3. Spark核心概念
在金融批处理场景中，Spark 之所以能成为主力计算引擎，核心在于它提供了一套从“底层弹性数据集”到“高层结构化查询”的完整抽象，并在背后用 Catalyst 与 Tungsten 把用户逻辑编译成高效的物理执行计划。理解这些概念之间的分工与协作，是后续用 Iceberg 做增量批处理、对账、跑批调度的基础。
一、四层抽象：从 RDD 到 Spark SQL
Spark 的数据抽象可以按“结构化程度”和“优化空间”两个维度来理解。RDD 是最底层的分布式对象集合，只描述“怎么算”，不携带 Schema，因此优化器几乎无法介入；DataFrame 与 Dataset 则携带 Schema，属于结构化 API，能进入 Catalyst 优化流程；Spark SQL 是面向 SQL 文本的入口，最终同样落到 DataFrame/Dataset 执行。
| 抽象 | 类型安全 | Schema | 优化能力 | 典型金融场景 |
|---|---|---|---|---|
| RDD | 编译期强类型 | 无 | 仅靠算子手工优化 | 非结构化日志清洗、自定义分区 |
| DataFrame | 运行时弱类型 | 有 | Catalyst + Tungsten | 交易流水聚合、宽表加工 |
| Dataset | 编译期强类型 | 有 | Catalyst + Tungsten | 带领域模型的账务对象处理 |
| Spark SQL | 弱类型（SQL） | 有 | Catalyst + Tungsten | 对账 SQL、监管报表 |
在 Scala 中，Dataset 可以看作 DataFrame 的强类型版本；在 Python 中，DataFrame 即 Dataset[Row]，没有单独的强类型 Dataset。金融工程里常见的做法是：接入层用 DataFrame 读 Iceberg 表，中间加工用 SQL 表达业务规则，只有需要复杂自定义逻辑时才回退到 RDD。
二、Catalyst 优化器：把“逻辑计划”变成“物理计划”
Catalyst 是 Spark SQL 的查询优化框架，基于 Scala 的模式匹配和规则引擎实现。它把用户写的 DataFrame 算子或 SQL 语句，依次经过四个阶段：
- 解析（Analysis）：借助 Catalog 把未解析的逻辑计划中的表名、列名绑定为具体元数据，生成已解析逻辑计划。
- 逻辑优化（Logical Optimization）：应用谓词下推、列裁剪、常量折叠、算子合并等规则，减少后续处理的数据量。
- 物理计划（Physical Planning）：把逻辑算子映射为可执行的物理算子，如 BroadcastHashJoin、SortMergeJoin、HashAggregate。
- 代价模型与选择：基于统计信息估算各物理计划代价，选出最优方案。
对金融批处理而言，最有价值的是“谓词下推”和“列裁剪”。例如按分区日期过滤时，Catalyst 会把过滤条件下推到 Iceberg 扫描层，只读取命中分区的 Parquet 文件，避免全表扫描。
// 逻辑计划中过滤条件下推示意
spark.sql("""
SELECT account_id, SUM(amount) AS total
FROM iceberg.finance.txn_detail
WHERE biz_date = '2024-06-30'
GROUP BY account_id
""").explain("formatted")
// 输出中可见 PushedFilters / PartitionFilters
// PartitionFilters: [isnotnull(biz_date), (biz_date = 2024-06-30)]
三、Tungsten 执行引擎：把 CPU 和内存用到极致
Catalyst 负责“选对路”，Tungsten 负责“跑得快”。Tungsten 是 Spark 的物理执行引擎，主要做三件事：
- 堆外内存管理：用二进制格式在堆外存储数据，减少 JVM 对象开销和 GC 压力。
- 全阶段代码生成（Whole-Stage Code Generation）：把多个算子融合成一段 Java 字节码，消除虚函数调用和中间对象。
- 缓存友好的数据布局：按列式或紧凑行式组织，提高 CPU 缓存命中率。
在金融跑批中，Tungsten 带来的收益非常直接：同样是对 10 亿行交易流水做分组聚合，开启代码生成后 CPU 时间可下降数倍，GC 停顿显著减少，批处理窗口更容易压进调度要求的时间片内。
判断是否触发 Tungsten 代码生成，可以看 explain 输出中是否出现 WholeStageCodegen 节点。若某段逻辑被标记为 Exchange 或 Sort 且未融合，通常意味着存在 Shuffle 或排序边界，需要关注数据倾斜与分区策略。
四、案例与误区复盘
假设某银行需要在 Iceberg 上做 T+1 账户日终余额汇总，输入是 txn_detail 流水表，输出是 acct_balance_daily。一个典型的执行链路是：
val txn = spark.table("iceberg.finance.txn_detail")
.filter($"biz_date" === "2024-06-30")
.select($"account_id", $"amount", $"dr_cr")
val bal = txn.groupBy("account_id")
.agg(sum(when($"dr_cr" === "C", $"amount").otherwise(0)).as("credit"),
sum(when($"dr_cr" === "D", $"amount").otherwise(0)).as("debit"))
bal.writeTo("iceberg.finance.acct_balance_daily")
.option("merge-schema", "true")
.append()
这段代码会依次经过 Catalyst 的解析、谓词下推、聚合优化，再由 Tungsten 生成融合代码执行。常见误区有三类：
- 用 RDD 做结构化聚合：放弃 Catalyst 优化，手写 groupByKey 容易导致 Shuffle 膨胀和 OOM。
- 忽略分区过滤：在 DataFrame 里用
to_date包裹分区列，导致分区裁剪失效，全表扫描。 - 盲目 collect 到 Driver：金融明细数据量大，collect 会把结果拉回单机，直接击穿 Driver 内存。
在 Iceberg 表上做批处理时，务必让过滤条件直接作用在分区列上，避免使用 cast、to_date、substr 等函数包裹分区列，否则 Catalyst 无法下推分区过滤，扫描量会成倍增加。
总结来看，RDD 提供最基础的分布式容错能力，DataFrame/Dataset 与 Spark SQL 提供结构化表达，Catalyst 负责逻辑与物理优化，Tungsten 负责高效执行。四者配合，才让 Spark 在金融批处理中既能表达复杂业务规则，又能在海量数据下保持可接受的跑批时长。