4、离线特征计算架构：基于Spark的离线特征ETL流程、分区策略、数据倾斜处理与任务调优
在金融特征平台中，离线特征计算承担着历史回溯、批量生成与周期性更新的核心职责。基于 Apache Spark 的 ETL 架构是当前工业界的主流选择。本章聚焦于离线特征 ETL 的完整流程、分区策略设计、数据倾斜的典型场景与解决方案，以及 Spark 任务的参数调优方法。
4.1 离线特征ETL流程
离线特征 ETL 通常遵循“数据接入 → 特征计算 → 特征存储”的三阶段流水线。每个阶段内部包含多个子步骤，确保特征数据的准确性、一致性与时效性。
核心流程节点：
- 数据接入层：从 Kafka / HDFS / Hive 读取原始交易、用户行为、风控事件等数据，进行 schema 校验与类型转换。
- 特征计算层：执行窗口聚合、滑动统计、跨表 Join、时序特征衍生等逻辑，生成宽表特征。
- 特征存储层：将计算结果写入 Hive 分区表或 HBase / Redis 在线存储，供模型训练与实时推理使用。
4.2 分区策略设计
合理的分区策略是离线特征计算性能的关键。金融场景下，特征表通常按 时间 和 业务维度 进行复合分区，以平衡数据扫描范围与文件数量。
| 分区维度 | 典型粒度 | 适用场景 | 注意事项 |
|---|---|---|---|
| 时间分区 | dt=yyyyMMdd | 日级特征回溯、T+1 批量计算 | 避免跨过多分区扫描；小文件问题 |
| 业务分区 | product_type / tenant_id | 多租户特征隔离、分业务线计算 | 分区数不宜过多（建议 < 500） |
| 哈希分区 | hash(user_id) % N | 数据倾斜严重时的重分布 | 需配合动态重分区使用 |
最佳实践：在金融风控特征场景中，推荐使用
dt + event_type 双层分区。例如 dt=20250301/event_type=transaction，既支持按天回溯，又能按事件类型裁剪数据量。
4.3 数据倾斜处理
数据倾斜是离线特征 ETL 中最常见的性能瓶颈。金融数据中，头部用户、热门产品、高频交易事件往往导致部分分区数据量远超平均水平，引发 OOM 或任务长时间挂起。
4.3.1 倾斜场景识别
- Join 倾斜：大表 Join 小表时，小表 key 分布不均；或两张大表 Join 时，部分 key 数据量过大。
- 聚合倾斜：group by 操作中，少数 key 对应的数据行数远高于其他 key。
- 窗口函数倾斜：partition by 字段值分布不均，导致单个分区数据量过大。
4.3.2 解决方案
| 倾斜类型 | 解决方案 | 实现方式 |
|---|---|---|
| Join 倾斜 | 广播小表 / 倾斜 key 单独处理 | spark.sql.autoBroadcastJoinThreshold 调大；或过滤倾斜 key 单独 Join 后 union |
| 聚合倾斜 | 两阶段聚合（加盐） | 先对 key 加随机前缀做局部聚合，再去前缀做全局聚合 |
| 窗口函数倾斜 | 扩大分区粒度 / 使用 bucketing | 将 partition by 字段与辅助字段组合；或预先对数据做 bucket 分桶 |
注意：加盐聚合虽然能缓解倾斜，但会引入额外的 shuffle 开销。建议仅在倾斜 key 数量较少（如 top 10）时使用，且盐值范围不宜过大（通常 10~100）。
4.4 任务调优
Spark 离线特征任务的调优需要从资源配置、Shuffle 优化、动态执行与数据格式四个维度展开。以下表格总结了关键参数及其推荐值。
| 调优维度 | 参数 | 推荐值 | 说明 |
|---|---|---|---|
| 资源配置 | spark.executor.memory | 8g ~ 32g | 根据数据量调整，避免 OOM |
| 资源配置 | spark.executor.cores | 4 ~ 8 | 过高会导致 IO 竞争 |
| Shuffle 优化 | spark.sql.shuffle.partitions | 200 ~ 1000 | 根据数据量动态调整，避免小文件 |
| Shuffle 优化 | spark.shuffle.file.buffer | 64k ~ 256k | 增大可减少磁盘 IO 次数 |
| 动态执行 | spark.dynamicAllocation.enabled | true | 根据负载自动调整 Executor 数量 |
| 数据格式 | spark.sql.parquet.compression.codec | snappy / zstd | 平衡压缩比与解压速度 |
调优案例：某金融风控特征表日增量约 2TB，原始使用默认 200 个 shuffle 分区，每个分区约 10GB，导致频繁 spill 到磁盘。将
spark.sql.shuffle.partitions 调整为 800，同时开启 spark.dynamicAllocation，任务耗时从 4.5 小时降至 1.8 小时，且无 OOM 发生。
4.4.1 代码级优化示例
以下代码片段展示了在 Spark 中处理离线特征 ETL 时，如何结合分区裁剪与广播 Join 来提升性能。
// 1. 分区裁剪：只读取当天数据
val df = spark.read
.option("basePath", "/data/features")
.parquet("/data/features/dt=20250301")
// 2. 广播小表：维度表小于 100MB 时自动广播
val dimDF = spark.read.parquet("/data/dim/product")
val result = df.join(broadcast(dimDF), Seq("product_id"), "left")
// 3. 两阶段聚合处理倾斜
val saltedDF = df.withColumn("salt", (rand() * 100).cast("int"))
.withColumn("salted_key", concat(col("user_id"), lit("_"), col("salt")))
val partialAgg = saltedDF.groupBy("salted_key", "event_type")
.agg(sum("amount").as("partial_sum"))
val finalAgg = partialAgg.withColumn("original_key",
split(col("salted_key"), "_")(0))
.groupBy("original_key", "event_type")
.agg(sum("partial_sum").as("total_amount"))
提示：在金融场景中，建议对特征计算任务开启
spark.sql.adaptive.enabled=true（Spark 3.0+），AQE 可以自动合并小分区、优化 Join 策略，显著减少手动调优工作量。