📚 课程目录第3章批处理架构详解

3. 批处理架构详解：基于Spark/Hive的离线特征计算流程、调度与分区策略

在金融特征平台中，批处理架构承担着历史特征回溯、大规模特征全量计算以及T+1离线特征生产的核心职责。本章将深入剖析基于Apache Spark与Apache Hive的离线特征计算流程、任务调度策略与数据分区设计，帮助你在金融级数据量下构建稳定、高效、可回溯的批处理特征生产线。

3.1 离线特征计算的整体架构

离线特征计算通常采用“分层加工”的Lambda架构中的批处理层。数据从原始业务日志、交易流水、用户画像等源系统流入，经过ETL清洗、特征加工、聚合计算，最终写入特征存储（如HBase、ClickHouse或特征服务API）。

  核心设计原则： 离线特征计算必须支持 时间旅行（Time Travel），即能够回溯到任意历史时间点重新计算特征，以满足模型回测与A/B实验的需求。

3.2 基于Spark/Hive的特征计算流程

一个典型的离线特征计算任务包含以下五个阶段，每个阶段都有明确的输入输出与优化要点：

阶段	技术选型	核心操作	输出产物
1. 数据接入	Spark Structured Streaming / Hive LOAD	从Kafka或HDFS读取原始日志，解析为结构化DataFrame	ODS层原始表（增量/全量）
2. 数据清洗	Hive SQL / Spark SQL	去重、空值填充、异常值过滤、字段类型统一	DWD层明细宽表
3. 特征计算	Spark Core / Hive UDF	窗口聚合（滑动窗口、累计窗口）、多表Join、复杂UDF调用	DWS层特征宽表
4. 特征聚合	Spark SQL / Hive	按实体ID+时间分区做聚合，生成最终特征向量	ADS层特征结果表
5. 数据导出	Spark JDBC / Bulk Load	将特征结果写入在线存储（HBase/ClickHouse）	在线特征快照

实战建议： 在特征计算阶段，优先使用Spark的 window() 函数实现滑窗聚合，避免使用Hive UDF处理复杂时间逻辑。Spark的钨丝执行引擎（Tungsten）对窗口算子有深度优化，性能可提升3-5倍。

3.3 调度策略：从定时触发到依赖编排

金融场景的离线特征计算对调度有严格要求：既要保证数据产出的时效性（通常在凌晨6:00前完成T-1特征），又要支持失败重跑与数据回溯。业界通常采用以下调度模式：

时间触发调度： 使用Crontab或调度平台（如Apache DolphinScheduler、Airflow）设置固定时间窗口（如每天02:00启动），适用于数据源稳定、计算量固定的场景。
依赖触发调度： 上游数据表分区就绪后自动触发下游任务，通过Hive Metastore的分区元数据判断依赖是否满足，避免空跑与资源浪费。
数据回溯调度： 支持指定历史日期范围批量重跑，调度系统自动生成多个并行实例，每个实例处理一个时间分区的特征计算。

3.4 分区策略：时间分区与业务分区

合理的分区设计是离线特征计算性能的基石。在金融特征平台中，我们通常采用两级分区策略：

一级分区（时间分区）： 按日期（dt=yyyy-MM-dd）或小时（hh=HH）分区，便于时间回溯与增量计算。Hive Metastore通过分区剪枝（Partition Pruning）快速定位数据范围。
二级分区（业务分区）： 按业务线（biz_type）、实体类型（entity_type）或数据等级（data_level）分区，实现计算隔离与权限控制。

分区层级	分区字段	示例值	主要作用
一级（时间）	`dt`	`2024-06-15`	支持时间旅行、增量计算、数据回溯
二级（业务）	`biz_type`	`loan`, `credit`, `fraud`	业务隔离、资源管控、并行调度
二级（实体）	`entity_type`	`user`, `merchant`, `device`	特征按实体分类存储，减少扫描范围

避坑指南： 避免使用 user_id 或 order_id 等高基数字段作为分区键。Hive分区数超过10万后，Metastore的元数据操作（如 SHOW PARTITIONS）会显著变慢，甚至导致OOM。建议将高基数字段作为Bucket（分桶）键，而非分区键。

3.5 代码示例：Spark离线特征计算任务

以下是一个完整的Spark离线特征计算代码片段，展示了从Hive读取ODS数据、计算滑动窗口特征、写入分区表的典型流程：

// SparkSession 初始化
val spark = SparkSession.builder()
  .appName("OfflineFeatureCompute")
  .config("spark.sql.shuffle.partitions", "200")
  .config("spark.sql.adaptive.enabled", "true")
  .enableHiveSupport()
  .getOrCreate()

// 1. 读取ODS层交易流水表（按dt分区）
val tradeDF = spark.sql(
  """
    |SELECT user_id, trans_amount, trans_time, dt
    |FROM ods.trade_log
    |WHERE dt = '${biz_date}'
  """.stripMargin
)

// 2. 定义滑动窗口：过去7天累计交易金额
import org.apache.spark.sql.expressions.Window
val windowSpec = Window.partitionBy("user_id")
  .orderBy("trans_time")
  .rowsBetween(-6, 0)  // 当前行及前6行（共7天）

val featureDF = tradeDF
  .withColumn("amt_7d_sum", sum("trans_amount").over(windowSpec))
  .withColumn("amt_7d_avg", avg("trans_amount").over(windowSpec))
  .withColumn("trans_cnt_7d", count("trans_amount").over(windowSpec))

// 3. 写入DWS层特征表（动态分区）
featureDF.write
  .mode("overwrite")
  .format("parquet")
  .partitionBy("dt", "biz_type")
  .saveAsTable("dws.user_trade_features_7d")

// 4. 触发计算并监控
spark.sql("MSCK REPAIR TABLE dws.user_trade_features_7d")
println(s"特征计算完成，日期分区: ${biz_date}")

性能调优要点： 对于金融级海量数据（单日百亿级记录），建议开启Spark动态资源分配（spark.dynamicAllocation.enabled=true）与AQE（自适应查询执行），并根据数据量调整 spark.sql.shuffle.partitions 为200-500之间，避免小文件过多。

3.6 批处理架构的金融级保障

在金融特征平台中，批处理架构还需要额外关注以下三个关键点：

数据一致性： 使用Hive ACID事务或Spark的 overwrite 模式确保分区写入的原子性，避免下游读到不完整的数据。
血缘追踪： 通过Atlas或自研血缘系统记录每个特征字段的来源表与计算逻辑，满足监管审计要求。
异常告警： 对特征产出延迟、数据量波动超过阈值（如环比下降50%）、空值率异常等场景设置实时告警，确保第一时间发现并修复。

通过以上架构设计与实践，基于Spark/Hive的离线批处理架构能够稳定支撑金融场景下千亿级特征数据的每日计算与回溯需求，为后续的实时特征与模型训练提供可靠的数据基础。