4、离线批处理管道：基于 Spark/Flink 的批量特征计算

聊完实时特征，咱们得回头看看另一条腿——离线批处理。说实话，很多公司一开始只做实时，觉得批处理太“土”。但我在几个大流量项目里踩过坑之后，才明白一个道理：没有离线兜底，实时就是空中楼阁。

为什么？你想想看，实时特征依赖的窗口数据，一旦任务重启或者数据回溯，历史状态全丢了。这时候，离线批处理就是那个“救火队员”。

4.1 批处理管道的定位：T+1 与小时级调度

离线批处理，说白了就是定时跑任务。最常见的两种节奏：

• T+1 调度：每天凌晨跑前一天的全量数据。适合用户画像、历史统计特征这类变化慢的指标。
• 小时级调度：每整点跑过去一小时的数据。适合需要近实时但允许几分钟延迟的场景，比如实时推荐的冷启动补充。

我个人习惯把这两种调度放在同一个管道里，用参数控制。比如：

# 伪代码：调度配置
batch_config = {
    "mode": "hourly",        # 可选 daily / hourly
    "window_size": "1h",     # 小时级窗口
    "trigger_time": "0 * * * *",  # Cron 表达式
    "backfill": True         # 是否支持回溯
}

嗯，这里要注意：回溯能力是离线管道的核心。我在项目中遇到过数据源延迟了 3 小时才到位的情况，如果没有回溯机制，那批特征就全废了。

4.2 基于 Spark 的批量特征计算

Spark 是我做离线批处理的首选。原因很简单：生态成熟，社区活跃，踩坑有人填。

一个典型的 Spark 批处理管道长这样：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count, window

spark = SparkSession.builder \
    .appName("FeatureBatch") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .getOrCreate()

# 读取 Iceberg 表
df = spark.read.format("iceberg").load("catalog.db.user_events")

# 特征聚合：按用户+小时窗口
feature_df = df.groupBy(
    col("user_id"),
    window(col("event_time"), "1 hour")
).agg(
    count("*").alias("event_count"),
    avg("amount").alias("avg_amount")
)

# 写入特征表
feature_df.write.format("iceberg") \
    .mode("overwrite") \
    .save("catalog.db.user_hourly_features")

这段代码看起来简单，但有几个坑我曾经踩过：

• 窗口对齐：Spark 的 window 默认按时间戳对齐，但如果你用 T+1 调度，要小心时区偏移。我建议统一用 UTC 时间。
• 数据倾斜：热门用户的 event_count 可能比其他用户高几个数量级。我一般用 salting 或者 repartition 来打散。
• 增量 vs 全量：小时级调度建议用增量读取（只读过去一小时的数据），避免全表扫描。Iceberg 的 time travel 在这里特别好用。

4.3 Flink 在批处理中的角色

很多人以为 Flink 只能做实时。其实 Flink 的 Batch 模式 在离线场景下也很能打。尤其是当你需要批流一体的时候，Flink 可以让你用同一套代码跑实时和离线。

我举个例子：

// Flink Batch 模式读取 Iceberg
TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inBatchMode());

tEnv.executeSql("CREATE TABLE user_events (" +
    "  user_id STRING," +
    "  event_time TIMESTAMP(3)," +
    "  amount DOUBLE" +
    ") WITH (" +
    "  'connector' = 'iceberg'," +
    "  'catalog-database' = 'catalog'," +
    "  'catalog-table' = 'user_events'" +
    ")");

Table result = tEnv.sqlQuery(
    "SELECT user_id, " +
    "  TUMBLE_END(event_time, INTERVAL '1' HOUR) AS window_end, " +
    "  COUNT(*) AS event_count, " +
    "  AVG(amount) AS avg_amount " +
    "FROM user_events " +
    "GROUP BY user_id, TUMBLE(event_time, INTERVAL '1' HOUR)"
);

result.execute().print();

你看，代码结构和实时模式几乎一样，只是把 inStreamingMode() 换成了 inBatchMode()。这就是批流一体的魅力。

4.4 数据湖集成：Iceberg / Delta Lake

离线批处理离不开数据湖。我最早用 Hive 表，后来换成 Iceberg，再后来也试过 Delta Lake。说说我的感受：

我个人更倾向 Iceberg，原因有三：

1. Flink 原生支持：Iceberg 的 Flink connector 是官方维护的，集成起来很顺滑。
2. 分区演进：Iceberg 允许你改分区策略而不重写数据。我在项目中遇到过业务方要求改分区字段，Iceberg 直接原地搞定。
3. 隐藏分区：Iceberg 自动管理分区，你不需要在 SQL 里显式指定 PARTITION BY。这对写 SQL 的分析师特别友好。

4.5 离线批处理管道架构图

下面这张图展示了离线批处理管道的完整流程：

4.6 实践中的关键点

最后，分享几个我在项目中总结的经验：

• 数据回溯是常态：不要假设数据源永远准时。我建议离线管道默认支持至少 7 天的回溯能力。
• 监控报警不能少：批处理任务失败后，如果没人发现，特征就会“断供”。我一般会加一个 data freshness 监控，如果特征表超过 2 小时没更新就报警。
• 测试环境要隔离：我曾经因为测试环境的批处理任务写入了生产特征表，导致线上模型用了脏数据。后来我强制要求所有离线任务必须带环境标签。