3. 数据采集与接入：数据源类型、接入方式与采样策略
特征流水线的第一道关卡是数据采集与接入。生产环境中，模型依赖的数据源类型多样、接入方式各异，且原始数据量往往远超实际训练所需。本章从数据源类型、数据接入方式、数据采样策略三个维度，系统梳理构建稳定、高效特征流水线的底层能力。
3.1 数据源类型
推荐系统与机器学习模型的特征数据通常来自三类核心数据源：日志、数据库、消息队列。三类数据源在时效性、存储结构、访问模式上存在显著差异，实际流水线往往需要混合使用。
| 数据源类型 | 典型载体 | 数据特点 | 常见用途 | 接入难点 |
|---|---|---|---|---|
| 日志 | Nginx 日志、应用日志、客户端埋点日志 | 半结构化/非结构化、追加写入、量大 | 用户行为序列、曝光日志、点击日志 | 格式解析、字段对齐、时间戳乱序 |
| 数据库 | MySQL、PostgreSQL、HBase、Redis | 结构化、支持随机读写、有事务保证 | 用户画像、物品元数据、离线统计特征 | 连接池管理、增量同步、慢查询 |
| 消息队列 | Kafka、RocketMQ、Pulsar | 流式、高吞吐、解耦生产与消费 | 实时特征、事件驱动、在线学习 | 消费偏移管理、消息回溯、顺序保证 |
实践建议： 在搭建特征流水线初期，优先从日志和数据库入手，待实时特征需求明确后再引入消息队列。避免一开始就追求全链路流式化，增加系统复杂度。
3.2 数据接入方式
根据数据处理时效性的不同，数据接入分为批处理和流处理两种范式。现代特征流水线通常采用 Lambda 架构或 Kappa 架构，将两者结合使用。
3.2.1 批处理（Batch Processing）
批处理适用于离线特征计算、全量数据回溯、模型训练样本生成等场景。典型工具包括 Apache Spark、Hive、MapReduce。
- 调度周期：小时级、天级、周级。
- 数据一致性：通过快照或时间窗口保证，支持重跑。
- 典型流程：从 HDFS / 数据仓库读取全量数据 → 特征变换 → 写入特征存储或训练样本表。
# 批处理特征接入伪代码（PySpark）
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("batch_feature_pipeline").getOrCreate()
# 从 Hive 读取用户行为表（T-1 全量）
df = spark.sql("SELECT user_id, item_id, action, timestamp FROM user_behavior WHERE dt = '2025-03-01'")
# 特征变换：统计用户近7天点击次数
from pyspark.sql.window import Window
import pyspark.sql.functions as F
window_spec = Window.partitionBy("user_id").orderBy("timestamp").rowsBetween(-6, 0)
df_features = df.withColumn("click_7d_cnt", F.count(F.when(F.col("action") == "click", 1)).over(window_spec))
# 写入特征表
df_features.write.mode("overwrite").parquet("hdfs://feature_store/user_click_7d_cnt.parquet")
3.2.2 流处理（Stream Processing）
流处理适用于实时特征计算、在线反作弊、动态排序等场景。典型工具包括 Apache Flink、Spark Streaming、Kafka Streams。
- 处理延迟：毫秒级到秒级。
- 状态管理：依赖状态后端（RocksDB、内存）维护窗口聚合。
- 典型流程：从 Kafka 消费实时事件 → 时间窗口聚合 → 输出到实时特征存储（如 Redis）。
-- 流处理特征接入伪代码（Flink SQL）
CREATE TABLE kafka_source (
user_id STRING,
item_id STRING,
action STRING,
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'user_behavior',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
);
CREATE TABLE redis_sink (
user_id STRING,
click_5min_cnt BIGINT,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'redis',
'redis.mode' = 'single',
'command' = 'SET'
);
INSERT INTO redis_sink
SELECT
user_id,
COUNT(*) AS click_5min_cnt
FROM kafka_source
WHERE action = 'click'
GROUP BY
user_id,
TUMBLE(event_time, INTERVAL '5' MINUTE);
避坑提醒： 流处理中的事件时间乱序是常见问题。务必设置合理的 watermark 和 allowed lateness，否则实时特征会出现偏差，直接影响模型在线效果。
3.3 数据采样策略
原始数据量通常远大于模型训练所需，直接全量接入会导致存储膨胀、计算成本飙升。数据采样策略在保证特征分布代表性的前提下，有效降低数据规模。
| 采样策略 | 方法描述 | 适用场景 | 注意事项 |
|---|---|---|---|
| 随机采样 | 按固定比例或固定行数随机抽取 | 数据分布均匀、无偏场景 | 对长尾特征可能采样不足 |
| 分层采样 | 按类别/特征值分层后按比例采样 | 类别不平衡、需要保证各层代表性 | 需提前定义分层字段 |
| 时间窗口采样 | 只保留最近 N 天/小时的数据 | 特征时效性强、旧数据价值衰减快 | 需监控窗口外数据分布漂移 |
| 负样本降采样 | 对负样本按比例丢弃，配合权重校正 | CTR/CVR 模型正负样本极度不平衡 | 训练时需对负样本加权还原分布 |
| 蓄水池采样 | 流式场景下等概率抽取固定大小样本 | 无法预知数据总量的流式接入 | 实现简单，但无法感知分布变化 |
采样策略选择原则：
- 离线训练场景优先使用分层采样，保证各特征域均有足够样本。
- 实时特征接入场景使用时间窗口采样 + 蓄水池采样组合，兼顾时效性与样本量控制。
- 负样本降采样后，务必在 loss 函数中引入样本权重，否则模型预估概率会系统性偏高。
3.4 数据接入流水线整体流程
将数据源、接入方式、采样策略串联起来，形成一条完整的特征数据接入流水线。以下流程图展示了从原始数据到特征存储的标准路径。
3.5 本章小结
- 数据源类型：日志、数据库、消息队列各有适用场景，混合使用是常态。
- 数据接入方式：批处理保证数据一致性与可回溯性，流处理提供低延迟实时特征。
- 数据采样策略：根据数据分布、时效性、样本平衡需求选择合适的采样方法，并配合权重校正。
- 流水线整体设计：从数据源到特征存储，每个环节都需要监控与反馈机制，确保特征质量稳定。