3. 数据 Ingestion 与验证：ExampleGen 组件详解、数据格式支持、数据统计与 Schema 生成

好，咱们进入第三章。说实话，数据 ingestion 这块，是整条 TFX 流水线的第一道关卡。你想想看，如果数据进都进不来，或者进来的是脏数据，后面模型训练得再好也是白搭。我在项目里见过太多人把精力全花在调模型上，结果数据一换，全崩了。嗯，咱们今天就把这第一关彻底打通。

3.1 ExampleGen 组件：流水线的入口

ExampleGen 说白了就是 TFX 的「数据搬运工」。它的任务很简单：把原始数据读进来，转成 TFX 内部统一的 tf.Example 格式，然后分发给下游组件。

我个人习惯把 ExampleGen 分成两类：

  • 默认 ExampleGen：针对常见格式（CSV、TFRecord、BigQuery 等），TFX 已经内置好了实现。
  • 自定义 ExampleGen：当你遇到奇葩数据源（比如私有数据库、特殊日志格式）时，需要自己写 Executor。

来看一个最常用的 CSV ExampleGen 示例：

from tfx.components import CsvExampleGen
from tfx.orchestration import pipeline

example_gen = CsvExampleGen(
    input_base='/path/to/data',
    output_config={
        'split_config': {
            'splits': [
                {'name': 'train', 'hash_buckets': 8},
                {'name': 'eval', 'hash_buckets': 2}
            ]
        }
    }
)

这里有个细节要注意：hash_buckets 参数决定了训练集和验证集的分割比例。8:2 是常见做法，但如果你数据量特别大，我建议用 9:1 甚至 99:1。为什么？因为验证集太大其实浪费，你想想看，100 万条数据里抽 10 万条做验证，和抽 1 万条，效果差别不大，但训练集少了 9 万条。

我的小技巧： 如果你数据是按日期分片的，可以用 input_config 里的 date_range 参数，按时间窗口切分数据。这样能避免未来数据泄露到训练集中。

3.2 数据格式支持：CSV vs TFRecord

TFX 原生支持的数据格式不少，但实际项目中用得最多的就两种：CSV 和 TFRecord。我分别说说它们的坑和优势。

3.2.1 CSV 格式

CSV 是最容易上手的格式。你只要把数据整理成表格，每列一个特征，每行一个样本，ExampleGen 就能自动解析。但注意：

  • 缺失值处理：CSV 里的空单元格，ExampleGen 默认会报错。你需要用 csv_args 参数指定缺失值标记，比如 na_value=''。
  • 类型推断：ExampleGen 会尝试自动推断每列的数据类型。但有时候它会搞错，比如把「用户ID」这种整数当成数值特征。我建议你后面用 Schema 显式指定类型。
example_gen = CsvExampleGen(
    input_base='/data',
    csv_args={
        'na_value': '',
        'quotechar': '"',
        'skip_blank_lines': True
    }
)

3.2.2 TFRecord 格式

TFRecord 是 TensorFlow 的亲儿子格式。它把数据序列化成二进制，读取速度比 CSV 快 3-5 倍。我在处理 TB 级数据时，一定会用 TFRecord。

但 TFRecord 有个门槛：你得自己写转换脚本。比如把 CSV 转成 TFRecord：

import tensorflow as tf

def csv_to_tfrecord(csv_path, tfrecord_path):
    with tf.io.TFRecordWriter(tfrecord_path) as writer:
        for row in pd.read_csv(csv_path, chunksize=1000):
            for _, record in row.iterrows():
                example = tf.train.Example(
                    features=tf.train.Features(
                        feature={
                            'feature1': tf.train.Feature(
                                float_list=tf.train.FloatList(value=[record['feature1']])
                            ),
                            'feature2': tf.train.Feature(
                                int64_list=tf.train.Int64List(value=[record['feature2']])
                            )
                        }
                    )
                )
                writer.write(example.SerializeToString())
我曾经踩过的坑： 写 TFRecord 时，每个 tf.train.Example 的大小不要超过 2MB。否则读取时会爆内存。我当年有个特征列是「用户行为序列」，一条记录就 10MB，结果流水线跑了一小时直接 OOM。后来我把长序列拆成多个小 Example，问题就解决了。

3.3 数据统计与 Schema 生成

数据进来之后，第一件事不是训练，而是「摸清家底」。你需要知道：数据长什么样？有没有异常值？特征分布是否合理？

TFX 提供了两个组件干这事：StatisticsGen 和 SchemaGen。

3.3.1 StatisticsGen：给数据做体检

StatisticsGen 会计算每个特征的统计量，比如均值、方差、缺失率、分位数等。输出是一个 DatasetStatistics 对象，你可以用 tfdv 可视化。

from tfx.components import StatisticsGen

statistics_gen = StatisticsGen(
    examples=example_gen.outputs['examples']
)

跑完之后，你可以用 tfdv.visualize_statistics() 看结果。我个人习惯重点关注三个指标：

  • 缺失率：超过 50% 缺失的特征，建议直接扔掉。
  • 异常值：比如「年龄」列出现 999 岁，明显是脏数据。
  • 分布偏度：如果某个特征的分布严重偏斜，后面可能需要做 log 变换。

3.3.2 SchemaGen：给数据定规矩

SchemaGen 会根据 StatisticsGen 的输出，自动推断出数据的 Schema。Schema 就是数据的「合同」——它规定了每个特征的类型、取值范围、是否允许缺失等。

from tfx.components import SchemaGen

schema_gen = SchemaGen(
    statistics=statistics_gen.outputs['statistics'],
    infer_feature_shape=True
)

生成的 Schema 是一个 protobuf 对象，长这样：

feature {
  name: "age"
  type: FLOAT
  presence {
    min_fraction: 1.0
  }
  shape {
    dim {
      size: 1
    }
  }
}

但自动生成的 Schema 不一定完美。比如它可能把「性别」列推断为整数，而实际上应该是枚举。这时候你需要手动修改 Schema：

import tensorflow_data_validation as tfdv

schema = tfdv.load_schema_text('path/to/schema.pbtxt')
tfdv.set_domain(schema, 'gender', 
                domain=tfdv.schema_pb2.IntDomain(min=0, max=1))
tfdv.write_schema_text(schema, 'path/to/updated_schema.pbtxt')
核心要点： Schema 是数据质量的「守门员」。一旦 Schema 定下来，后续的 ExampleValidator 组件就会用它来检测新数据是否合规。如果新数据里出现了 Schema 没定义的特征，或者某个特征的值超出了范围，流水线会直接报错。这能帮你提前发现数据漂移。

3.4 实战建议：数据 ingestion 的黄金流程

根据我多年的经验，一个稳健的数据 ingestion 流程应该包含以下步骤：

  1. 数据采样：先用 1% 的数据跑一遍流水线，确保 ExampleGen 能正常工作。
  2. 生成统计：用 StatisticsGen 看数据分布，发现明显异常。
  3. 定义 Schema：基于统计结果，手动调整 Schema，特别是特征类型和取值范围。
  4. 验证 Schema：用 ExampleValidator 检查全量数据是否符合 Schema。
  5. 正式运行：确认无误后，再跑全量数据。

嗯，这套流程看起来多了一步，但能帮你省下大量 debug 时间。我见过太多人跳过前四步，直接全量跑，结果跑了两小时发现数据格式不对，又得重来。何必呢？

好了，第三章就到这里。下一章咱们聊聊数据验证和异常检测——说白了，就是怎么让流水线自己发现数据变坏了。

← 上一章 📖 返回目录 下一章 →