这可能看起来很愚蠢，但这是我在这里的第一篇文章。很抱歉做错了什么。

我目前正在使用 python2.7 构建一个带有 TFX 0.11（即 tfdv-tft-tfserving）和 tensorflow 1.11 的简单 ML 管道。我目前有一个 apache-flink 集群，我想将它用于 TFX。我知道 TFX 背后的框架是 apache-beams 2.8，它 (apache-beams) 目前通过可移植的运行器层支持使用 python SDK 的 flink。

但问题是我如何通过这个便携式运行器概念使用带有 flink 运行器的 apache-beams 在 TFX (tfdv-tft) 中编码，因为 TFX 目前似乎只支持 DirectRunner 和 DataFlowRunner (Google Cloud)。

我在网上搜索了一段时间，看到TFX网站的最后一行，

“请使用 tensorflow-transform 标签将有关使用 tf.Transform 的任何问题直接提交给 Stack Overflow。”

这就是我在这里的原因。任何想法或解决方法都非常感谢。谢谢！

我正在尝试使用 tensorflow-transform v0.11.0 预处理大量数据（一个 tfrecord 文件 ~1Go）并仅在本地传输。

我的代码很大程度上来自https://github.com/tensorflow/transform/blob/master/examples/census_example.py

我有一个适用于较小数据集（<100Mo）的光束管道，但随着我添加更多数据，处理时间会急剧增加。作为 tf-transform 和 apache Beam 的新手，我很难找到问题的原因和解决方案……而且我想避免使用 google DataFlow。

如果我理解正确的话，我的管道使用 beam directRunner 在本地运行，但它只使用一个核心。使用多核可能是缩短预处理时间的一种方法，但我不知道使用 beam directRunner 是否可行。有没有办法让张量流转换管道在我机器上的多个核心上运行？

我查看了光束管道和 directRunner 的选项，但找不到任何关于让运行器访问多个核心或为管道创建多个 directRunner 的迹象。

非常感谢您为我提供的任何帮助！

概述

我按照以下指南编写 TF 记录，我曾经tf.Transform在其中预处理我的特征。现在，我想部署我的模型，为此我需要将此预处理功能应用于真实的实时数据。

我的方法

首先，假设我有 2 个功能：

我有transform_fn来自 Apache Beam 的，居住在working_dir=gs://path-to-transform-fn/

然后我使用以下方法加载转换函数：

tf_transform_output = tft.TFTransformOutput(working_dir)

我认为在生产中服务的最简单方法是获取一个处理过的数据的 numpy 数组，然后调用model.predict()（我使用的是 Keras 模型）。

为此，我认为transform_raw_features()方法正是我所需要的。

但是，似乎在构建架构之后：

我得到：

现在，我假设发生这种情况是因为我tf.VarLenFeature()在preprocessing_fn.

我使用以下方法构建元数据：

简而言之，给定一本字典：

d = {'amount': [50], 'age': [32]}，我想应用这个transform_fn，并适当地缩放这些值以输入到我的模型中进行预测。这个字典正是函数PCollection处理数据之前的我的格式pre_processing()。

管道结构：

最后ParDo()是：

我正在尝试运行 TensorFlow Transform，使用 Python、Apache Flink 作为 Beam Runner。我注意到 Beam 没有 AWS S3 作为 io 连接器，并且想知道任何解决方法。

这是支持的 io 连接器列表，但 Python+S3 甚至不在路线图中。

我可以想到两种解决方法：

1. 将 S3 存储桶作为本地驱动器挂载到 EC2 实例
2. 使用他们的指南编写我自己的 Python S3 连接器。

我想知道是否还有其他创造性（简单）的出路。

谢谢！

我在这里运行 Sentiment 示例以进行 tensorflow 转换。 https://github.com/tensorflow/transform/blob/master/examples/sentiment_example.py

对于第 78-98 行中定义的 fn ReadAndShuffleData()，是否有可能以类似的方式加载文件但从 HDFS，而不是 GCS？

我已经用几个梁 API（beams-2.8.0）尝试了一整天但失败了，我认为最有希望的是使用 beams.io.hadoopfilesystem。但是这个 fn 实际上会产生一个 python 文件对象，并且无法在梁管道中使用 beams.io.ReadFromText() 读取。

我还正确地传入了 HadoopFileSystemPipelineOptions。任何人都可以告诉我解决问题的方向或 2/3 行代码片段或解决方法？非常感谢！

ps hadoop 2.7.7，beams 2.8，数据加载正确。

我想我可能在这里缺乏一些理论理解，任何参考将不胜感激！

我在使用带有 export_saved_model 函数的估计器进行训练后，使用 tensorflow 转换导出模型的所有示例。我正在使用不使用估计器的现有训练代码并使用 saved_model.simple_save 保存。

在不使用估计器的情况下保存带有服务模型的 transform_fn 的最佳实践方法是什么？

我们可以使用 pip 安装 tensorflow-transform 但如何使用 Conda 安装它？

我尝试过使用conda install -c anaconda tensorflow-transform它的所有变体，但没有结果。

这个问题在周末开始出现。出于某种原因，感觉是一个 DataFlow 问题。

以前，我能够执行脚本并写入 TF 记录就好了。但是，现在，我无法初始化计算图来处理数据。

回溯是：

我正在使用 tensorflow==1.13.1 和 tensorflow-transform==0.9.0 和 apache_beam==2.7.0

tf transform 对于特征处理很方便，但在没有分布式计算的情况下在大型数据集上运行效率不高。tf transform 在 beam 上运行，据我了解，它可以使用多个运行器，如 dataflow、spark runner 等，但我找不到任何关于在 spark 上运行 tf transform 的示例。我想知道目前是否支持它。

如何使用 tensorflow-transform 来降低图像的平均值并使其在服务期间也能正常工作