📚 课程目录第4章 Spark批处理编程

4. Spark批处理编程

在金融批处理场景中，Spark 的核心价值不只是“能算”，而是“算得稳、算得快、算得可追溯”。本章聚焦批处理编程的五个关键抓手：DataFrame API、Spark SQL、列式文件读写、分区与分桶、缓存与持久化。它们共同决定了作业的执行计划、I/O 成本、Shuffle 规模与资源占用。

一、核心原理：DataFrame 与 Spark SQL 的统一执行模型

DataFrame 是带 Schema 的分布式数据集，本质是逻辑计划 + 表达式树。Spark SQL 则把 SQL 文本解析为同一套逻辑计划，再经 Catalyst 优化器生成物理计划。两者最终都落到 RDD 的 DAG 上执行，因此性能差异主要来自优化器能否下推过滤、裁剪列、合并分区。

金融数据常见宽表（如客户资产、交易流水、持仓明细），使用 DataFrame API 做列裁剪与谓词下推，比直接写 RDD 更易获得优化收益。例如读取 Parquet 时只选择必要列，Spark 会跳过无关列块，减少 I/O。

val df = spark.read.parquet("/warehouse/trade/dt=2024-06-01")
  .filter($"trade_amt" > 0)
  .select("cust_id", "trade_amt", "trade_time")
df.createOrReplaceTempView("trade_tmp")
spark.sql("""
  SELECT cust_id, SUM(trade_amt) AS amt
  FROM trade_tmp
  GROUP BY cust_id
""").show()

Catalyst 优化器会依次应用：谓词下推、列裁剪、常量折叠、Join 重排序、分区裁剪。理解这些规则，才能判断为什么同一份 SQL 写法不同，执行时间可能相差数倍。

二、方法流程：列式文件读写与分区分桶

Parquet 与 ORC 是列式存储，适合金融批处理的宽表扫描与聚合；CSV 可读性好但无 Schema、无压缩优势，通常只用于交换或小规模导入。写入时建议统一使用 Parquet，并按业务日期分区。

格式	Schema	压缩/编码	适用场景
Parquet	自带	列式压缩、字典编码	宽表聚合、长期存储
ORC	自带	列式压缩、索引	Hive 生态、高压缩比
CSV	需推断	文本压缩	交换、小批量导入

分区（Partition）按目录切分数据，如 dt=2024-06-01/，使过滤条件能直接裁剪目录；分桶（Bucket）按哈希把同一键的数据写入固定数量文件，常用于 Join 时避免 Shuffle。金融场景中，交易流水按 dt 分区、按 cust_id 分桶，可显著降低客户维度关联成本。

df.write
  .mode("overwrite")
  .partitionBy("dt")
  .bucketBy(64, "cust_id")
  .sortBy("cust_id")
  .saveAsTable("dwd_trade_bucket")

分区字段应选择高基数但查询中高频过滤的列；分桶数建议为集群核心数的 2～4 倍，且需与 Join 键一致才能触发 Bucket Join。

三、缓存与持久化：内存与磁盘的权衡

缓存（cache）与持久化（persist）用于复用中间结果。DataFrame 默认 MEMORY_AND_DISK，RDD 可指定 MEMORY_ONLY、DISK_ONLY 等。金融批处理中，若同一份宽表被多次聚合或多次 Join，缓存可避免重复扫描与 Shuffle。

val base = spark.read.parquet("/warehouse/cust_asset")
  .filter($"dt" === "2024-06-01")
  .select("cust_id", "asset_amt", "risk_level")
base.persist(StorageLevel.MEMORY_AND_DISK)
base.count()  // 触发缓存
// 后续多次使用 base 不再重复读取

缓存不是越多越好。缓存会占用 Executor 内存，可能引发 GC 或 spill；若中间结果只使用一次，缓存反而增加开销。使用后应调用 unpersist() 释放。

四、案例与复盘：金融批处理作业的典型误区

某银行日终持仓汇总作业，原始写法直接读取 CSV 并多次 groupBy，耗时 40 分钟。优化后改为：Parquet 列裁剪 + 按 dt 分区 + 对客户维表分桶 + 中间结果缓存，耗时降至 9 分钟。关键改动并非“换 API”，而是让优化器能下推、让 Shuffle 能减少。

误区一：用 CSV 做宽表长期存储，导致每次全量解析与 Schema 推断。
误区二：分区字段基数过低（如按性别分区），目录数量少但单目录过大，裁剪失效。
误区三：分桶键与 Join 键不一致，无法触发 Bucket Join，仍走 Shuffle。
误区四：缓存后未释放，多个作业共享 Executor 时内存争抢。

综上，Spark 批处理编程的优化主线是：用 DataFrame/SQL 让 Catalyst 有机会优化，用 Parquet/ORC 降低 I/O，用分区与分桶减少扫描与 Shuffle，用缓存复用中间结果，并在使用后及时释放。把这四点落到金融日终、风控跑批、监管报送等场景中，才能把“能跑”变成“跑得稳、跑得快”。