将旧版 ML 管道迁移到 TFX答案

【问题标题】：Migrating legacy ML pipeline to TFX将旧版 ML 管道迁移到 TFX
【发布时间】：2021-01-22 15:45:50
【问题描述】：

我们正在研究将我们的 ML 流水线从一组手动步骤转换为 TFX 流水线。不过，我确实有一些问题想了解一些额外的见解。

我们通常执行以下步骤（用于图像分类任务）：

加载图像数据和元数据
根据元数据过滤掉“坏”数据
确定基于图像的统计信息（Python 中的经典图像处理）：
1. 图像级别特征
2. 图像区域特征（区域根据微调的 EfficientDet 模型确定）
根据图像统计过滤掉“不良”数据
从此图像和元数据生成 TFRecords
过采样某些 TFRecord 以实现类平衡（使用 tf.data）
训练图像分类器
…

现在，我正在尝试将其映射到典型的 example TFX pipeline。

然而，这引发了一些问题：

我看到两个选项：
- ExampleGen 使用一个 CSV 文件，其中包含指向要加载的图像和要加载的元数据的指针（在步骤“1”之上）。然而：
  - 如果此 CSV 文件包含图像文件的路径，ExampleGen 是否可以加载图像数据并将其添加到其输出中？
  - ExampleGen 的输出是流式输出，还是所有示例数据的转储？
- ExampleGen 将 TFRecords 作为输入（上述步骤“5”的输出）
  
  -> 这意味着我们仍然需要在 TFX 之外实施步骤 1-5……这会降低 TFX 对我们的价值……
您能否建议最好的前进方式？
StatisticsGen 是否还可以生成基于每个示例的统计信息（例如基于经典图像处理的某些图像（区域）特征）？还是应该在 ExampleGen 中实现？还是……？
可以使用元数据存储缓存计算的统计信息吗？如果是，是否有可用的示例？

使用经典图像处理计算基于图像的特征很慢。如果有新数据可用，触发 TFX 输入组件执行，理想情况下应该从缓存中加载已计算的统计信息。
ExampleValidator 可能会拒绝某些示例（例如缺失数据、异常值……）是否正确？
如何在此设置中实现网络输入端的类平衡（不是通过损失函数）（通常我们通过使用 tf.data 对 TFRecord 进行过采样来实现）？如果这是在 ExampleGen 级别完成的，那么 ExampleValidator 可能仍会拒绝一些可能再次不平衡数据的示例。对于大数据 ML 任务，这似乎不是一个大问题，但对于小数据 ML 任务（通常在医疗保健环境中就是这种情况）而言，它变得至关重要。所以我希望在 Transform 组件之前有一个 TFX 组件，但是这个块应该可以访问所有数据，而不是以流的方式（参见我之前关于 ExampleGen 输出的问题）......

感谢您的见解。

【问题讨论】：

标签： tfx

【解决方案1】：

我将尝试用我在 tfx 方面的经验来解决大多数问题。

我有一个数据流作业，我运行它来预处理我的图像、标签、特征等，并将所有这些转换为 tfrecord。它位于 tfx 之外，仅在数据刷新时运行。

您也可以这样做，这是一个非常简单的代码 sn-p，我用它来调整我所有的图像大小并创建简单的特征。

try:
      image = tf.io.decode_jpeg(image_string)
      image = tf.image.resize(image,[image_resize_size,image_resize_size])
      image = tf.image.convert_image_dtype(image/255.0, dtype=tf.uint8)
      image_shape = image.shape

      image = tf.io.encode_jpeg(image,quality=100)

      feature = {
        'height': _int64_feature(image_shape[0]),
        'width' : _int64_feature(image_shape[1]),
        'depth' : _int64_feature(image_shape[2]),
        'label' : _int64_feature(labels_to_int(element[2].decode())),
        'image_raw' : _bytes_feature(image.numpy())
      }

      tf_example = tf.train.Example(features=tf.train.Features(feature=feature))
    except:
      print('image could not be decoded')
      return None

获得 tfrecord 格式的数据后，我使用 ImportExampleGen 组件将数据加载到我的 tfx 管道中。接下来是 StatisticsGen，它将计算特征的统计信息。

在云中运行所有这些时，它在批处理模式下使用数据流。

您的元数据存储仅缓存管道元数据，但您的数据缓存在 gcs 存储桶中，元数据存储知道这一点。因此，当您重新运行管道时，如果您将缓存设置为 True，如果数据未更改，则不会重新运行 ImportExampleGen、StatisticsGen、SchemaGen、Transform。这在时间和成本方面具有巨大的优势。
ExampleValidator 输出一个工件，让您知道您的数据中有哪些数据异常。我创建了一个自定义组件来接收 examplevalidator 工件，如果我的数据不符合某些标准，我会通过在该组件中抛出错误来终止管道。我希望有一个组件可以停止管道，但我还没有找到，所以我的解决方法是抛出一个错误，阻止管道进一步前进。
通常当我创建一个 tfx 管道时，它是为了自动化机器学习过程。至此，我们已经完成了类平衡、特征选择等工作。因为这更多地属于预处理阶段。

我想您可以从技术上创建一个自定义组件，该组件接受一个 StatisticsGen 工件，对其进行解析并尝试进行一些类平衡，并创建一个带有 balances 类的新数据集。但老实说，我认为最好在预处理阶段进行。

【讨论】：