4. 样本生成流程：特征回溯、特征拼接、样本标注、样本存储

样本生成是特征平台连接离线特征与模型训练的桥梁。一个完整的样本生成流程包含四个核心阶段：特征回溯（获取历史特征）、特征拼接（关联标签与特征）、样本标注（生成监督信号）和样本存储（序列化与分区）。本章将逐一拆解每个阶段的技术细节与校验要点。

4.1 特征回溯：时间旅行与快照还原

特征回溯的核心能力是“回到过去某个时间点，获取当时应有的特征值”。这要求特征存储支持时间旅行（Time Travel）或增量快照（Snapshot）机制。

4.1.1 回溯实现方式对比

4.1.2 回溯校验要点

• 时间对齐：确保回溯时间戳与事件时间戳在同一时区，避免因时区偏移导致特征错位。
• 空值处理：回溯不到特征时，需明确填充策略（默认值 / 前向填充 / 丢弃样本）。
• 版本一致性：同一批次样本应使用同一特征版本回溯，避免因特征更新导致样本间不一致。

4.2 特征拼接：多源特征关联

特征拼接是将回溯后的特征与样本主键（如 user_id、item_id）进行关联，形成宽表。拼接方式分为点查拼接和批量拼接两种模式。

4.2.1 拼接模式对比

4.2.2 拼接校验清单

• 主键唯一性：检查拼接后主键是否出现重复，重复会导致样本膨胀或标签错位。
• 特征覆盖度：统计每个特征的 null 比例，若某特征 null 率超过阈值（如 30%），需排查回溯或拼接逻辑。
• 类型一致性：确保拼接后的特征类型与模型输入类型一致（如 int64 转 float32）。

4.3 样本标注：标签生成与时间窗口

样本标注是根据业务目标生成监督信号（标签）。常见标注方式包括固定窗口标注、事件触发标注和多任务标注。

4.3.1 标注方式对比

4.3.2 标注校验要点

• 标签泄漏：确保标签窗口不包含未来信息。例如，用 T+7 天的转化作为标签，但特征回溯不能包含 T+7 天的数据。
• 标签分布：统计正负样本比例，若严重失衡（如正样本 < 1%），需考虑采样或加权策略。
• 时间截断：对于窗口未结束的样本，需明确处理方式（丢弃 / 标记为未观测）。

4.4 样本存储：TFRecord / Hive / Parquet

样本生成完成后，需要选择合适的存储格式。不同格式在读取性能、压缩比和生态兼容性上差异显著。

4.4.1 存储格式对比

4.4.2 存储校验要点

• 分区策略：按日期/小时分区，避免单分区数据量过大（建议每分区 < 200GB）。
• Schema 校验：存储前校验字段名、类型、nullable 属性是否与预期一致。
• 数据完整性：对比存储前后的样本总数、特征均值、标签分布，确保无数据丢失。

4.4.3 样本存储代码示例

# TFRecord 写入示例（Python）
import tensorflow as tf

def serialize_sample(features, label):
    feature_dict = {
        'user_id': tf.train.Feature(int64_list=tf.train.Int64List(value=[features['user_id']])),
        'item_id': tf.train.Feature(int64_list=tf.train.Int64List(value=[features['item_id']])),
        'click_rate': tf.train.Feature(float_list=tf.train.FloatList(value=[features['click_rate']])),
        'label': tf.train.Feature(int64_list=tf.train.Int64List(value=[label])),
    }
    example = tf.train.Example(features=tf.train.Features(feature=feature_dict))
    return example.SerializeToString()

# 写入 TFRecord 文件
with tf.io.TFRecordWriter('train_samples.tfrecord') as writer:
    for sample in sample_batch:
        serialized = serialize_sample(sample['features'], sample['label'])
        writer.write(serialized)

# Parquet 写入示例（PySpark）
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("sample_generation").getOrCreate()

df = spark.createDataFrame([
    (1001, 2001, 0.35, 1),
    (1002, 2002, 0.12, 0),
], schema=['user_id', 'item_id', 'click_rate', 'label'])

df.write.mode("overwrite").parquet("hdfs://data/train_samples/dt=2024-01-15")

4.5 样本生成全流程校验清单