4、数据预处理与特征工程：Transform组件详解、TFT基础、特征工程最佳实践

好，咱们进入第四讲。说实话，很多同学觉得TFX里最神秘的就是Transform组件。我刚开始接触时也这么想——不就是数据预处理吗？搞这么复杂干嘛？后来在项目中踩了坑才明白，Transform组件是整个流水线的灵魂。它解决了生产环境中一个核心痛点：训练和推理时的数据预处理逻辑必须完全一致。

4.1 为什么需要Transform组件？

你想想看，训练时你做了归一化、分桶、词汇表映射。上线推理时，这些逻辑得一模一样吧？手动复制代码？我试过，维护成本高得吓人。有一次模型效果突然变差，查了两天才发现是训练和推理的归一化参数没对齐。

Transform组件就是来解决这个问题的。它把预处理逻辑定义成一张「蓝图」，训练时生成转换函数，推理时直接复用。说白了，一次定义，到处运行。

核心价值：Transform组件确保训练和serving的数据处理逻辑完全一致，避免「训练-推理偏差」。

4.2 TFT（TensorFlow Transform）基础

TFT是Transform组件的底层引擎。它基于Apache Beam实现，能处理大规模数据。我个人习惯把TFT看作一个「数据转换编译器」——你写的是高层逻辑，它帮你生成高效的执行计划。

4.2.1 核心概念：preprocessing_fn

这是TFT的灵魂函数。你在这个函数里定义所有预处理逻辑，TFT会自动分析数据、计算统计量、生成转换图。

import tensorflow_transform as tft

def preprocessing_fn(inputs):
    """输入是一个特征字典，输出也是特征字典"""
    outputs = {}
    
    # 数值特征：归一化到[0,1]
    outputs['age_scaled'] = tft.scale_to_0_1(inputs['age'])
    
    # 类别特征：生成词汇表
    outputs['city_vocab'] = tft.compute_and_apply_vocabulary(
        inputs['city'],
        top_k=100,  # 只保留前100个高频词
        num_oov_buckets=10  # 未知词分到OOV桶
    )
    
    # 文本特征：TF-IDF
    outputs['text_tfidf'] = tft.tfidf(
        inputs['text'],
        vocab_size=5000
    )
    
    return outputs

避坑指南：我曾经在preprocessing_fn里直接用了pandas的groupby操作，结果在Beam分布式环境下跑崩了。记住，preprocessing_fn里只能用TFT提供的算子，不能用pandas或numpy。

4.2.2 分析器（Analyzers）与转换器（Transformers）

TFT把操作分为两类：

类型 特点 常见算子
分析器 需要扫描全量数据，计算统计量 min, max, mean, variance, vocabulary
转换器 逐行处理，不依赖全局统计 scale_by_min_max, apply_vocabulary, bucketize

为什么会这样区分？因为分析器需要一次完整的数据扫描，计算量很大。而转换器是逐行处理的，可以流式执行。我建议你把分析器操作尽量合并，减少数据扫描次数。

4.3 Transform组件实战

好，理论说完了，咱们看看Transform组件在TFX流水线里怎么用。

4.3.1 组件配置

import tfx.v1 as tfx

transform = tfx.components.Transform(
    examples=example_gen.outputs['examples'],
    schema=schema_gen.outputs['schema'],
    module_file='/path/to/transform_module.py',  # 包含preprocessing_fn的文件
    materialize=True  # 是否物化转换后的数据
)

这里有个关键参数materialize。如果设为True，Transform会把转换后的数据保存到文件系统。我一般建议开启，因为后续的Trainer组件可以直接读取处理好的数据，省去重复计算。

4.3.2 完整示例：用户画像特征工程

我在做用户画像项目时，用到了下面这个预处理流程：

def preprocessing_fn(inputs):
    outputs = {}
    
    # 1. 处理缺失值
    # 数值型用中位数填充，类别型用'UNKNOWN'
    outputs['age'] = tft.fill_in_missing(inputs['age'], default_value=30)
    outputs['gender'] = tft.fill_in_missing(
        inputs['gender'], 
        default_value='UNKNOWN'
    )
    
    # 2. 数值特征：对数变换 + 归一化
    # 收入分布通常偏态，先取对数再归一化效果更好
    income_log = tf.math.log1p(inputs['income'])
    outputs['income_scaled'] = tft.scale_to_z_score(income_log)
    
    # 3. 时间特征：提取周期性特征
    # 用户活跃时间有周期性，用sin/cos编码
    hour = inputs['active_hour']
    outputs['hour_sin'] = tf.math.sin(2 * np.pi * hour / 24)
    outputs['hour_cos'] = tf.math.cos(2 * np.pi * hour / 24)
    
    # 4. 交叉特征：年龄分组与城市组合
    age_bucket = tft.bucketize(inputs['age'], num_buckets=5)
    city_id = tft.compute_and_apply_vocabulary(inputs['city'])
    # 手动做特征交叉
    outputs['age_city_cross'] = age_bucket * 1000 + city_id
    
    return outputs

注意：特征交叉时要注意数值范围。我遇到过交叉后特征值过大导致模型不收敛的情况。建议用哈希或者分桶后再交叉，控制特征空间大小。

4.4 特征工程最佳实践

做了这么多项目，我总结了几条铁律：

4.4.1 特征选择三原则

  1. 相关性优先：先做相关性分析，剔除冗余特征。我习惯用Spearman相关系数，对非线性关系也敏感。
  2. 覆盖率检查：特征缺失率超过50%的，直接扔掉。别想着填充，填充本身就是引入噪声。
  3. 稳定性验证：训练集和验证集的特征分布要一致。用PSI（群体稳定性指标）量化，PSI > 0.2的特征要小心。

4.4.2 类别特征处理

类别特征的处理，我踩过不少坑。简单总结一下：

  • 高基数特征（如用户ID）：用哈希分桶，或者做target encoding。别直接one-hot，维度爆炸。
  • 低基数特征（如性别）：直接用词汇表映射，OOV桶设小一点。
  • 文本特征：先用TFT的tfidf，效果不够再用预训练embedding。

我的习惯：对于类别特征，我一般先做频率统计。出现次数少于100次的，统一归为'RARE'类别。这样能有效控制词汇表大小，同时保留低频信息。

4.4.3 数值特征处理

数值特征看似简单，其实门道很多：

  • 偏态分布：先做对数变换或Box-Cox变换，再归一化。
  • 异常值：用分位数截断，我一般取1%和99%分位数。
  • 分桶：等宽分桶容易受异常值影响，建议用等频分桶。
# 异常值截断示例
def clip_outliers(x, lower_percentile=0.01, upper_percentile=0.99):
    lower = tft.quantiles(x, [lower_percentile])[0]
    upper = tft.quantiles(x, [upper_percentile])[0]
    return tf.clip_by_value(x, lower, upper)

4.4.4 时间特征工程

时间特征容易被忽视，但往往能带来显著提升：

  • 周期性编码：小时、星期、月份用sin/cos编码。
  • 滑动窗口统计：过去7天的均值、最大值、标准差。
  • 时间差特征：距离上次行为的时间间隔。

嗯，这里要注意：滑动窗口统计在TFT里实现比较麻烦，因为需要排序。我一般是在ExampleGen阶段用Beam的窗口函数预处理，或者放到Trainer里用tf.data处理。

4.5 性能优化技巧

Transform组件处理大数据时，性能是个大问题。我分享几个优化经验：

  1. 减少分析器数量：每个分析器都要扫描一次数据。能合并的尽量合并。
  2. 使用CombinedStats：TFT 0.30+支持合并统计量计算，能减少扫描次数。
  3. 调整Beam参数：增大worker数量，调整sharding策略。
# 合并统计量计算示例
def preprocessing_fn(inputs):
    # 错误做法：三个分析器，扫描三次
    # min_val = tft.min(inputs['age'])
    # max_val = tft.max(inputs['age'])
    # mean_val = tft.mean(inputs['age'])
    
    # 正确做法：一次扫描计算多个统计量
    stats = tft.combined_stats(inputs['age'], 
                               [tft.min, tft.max, tft.mean])
    outputs['age_scaled'] = (inputs['age'] - stats[2]) / (stats[1] - stats[0])

好了，这一讲的内容就到这里。Transform组件是TFX流水线的核心，掌握了它，你就能构建出健壮、可维护的数据预处理流程。下一讲我们会进入Trainer组件，看看怎么用这些处理好的数据训练模型。

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