上一篇解决了 Spark 的容错机制。这一篇进入 SQL 引擎的核心——Catalyst 优化器。

Spark SQL 不只是"在 Spark 上跑 SQL"。它是一个完整的查询编译器，把 SQL 文本或 DataFrame API 调用翻译成优化后的物理执行计划。这个翻译过程由 Catalyst 优化器驱动，经过四个阶段：解析、分析、优化、物理计划生成。

本文只抓一个问题：一条 SQL 从文本到可执行的物理算子树，中间经历了什么变换。

四阶段流水线

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
SQL 文本 / DataFrame API
      │
      ▼
┌─────────────────┐
│  ① Parsing      │  SQL 文本 → Unresolved Logical Plan
│  (ANTLR 解析器)  │  列名和表名尚未绑定到实际 schema
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  ② Analysis     │  Unresolved → Resolved Logical Plan
│  (Catalog 查询)  │  绑定列名、表名、数据类型
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  ③ Optimization │  Resolved → Optimized Logical Plan
│  (规则重写)      │  谓词下推、列裁剪、常量折叠...
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  ④ Planning     │  Optimized Logical → Physical Plan
│  (策略选择)      │  选择 Join 算法、Scan 方式、排序策略
└─────────────────┘

DataFrame API 不经过 Parsing 阶段——它直接构建 Resolved Logical Plan（因为列名和类型在 API 调用时已经确定）。这也是 DataFrame 和 SQL 在底层共享同一套优化器的基础。

Tree + Rule 架构

Catalyst 的核心设计是两个抽象：TreeNode 和 Rule。

TreeNode 是所有计划节点的基类。LogicalPlan、Expression、SparkPlan（物理计划）都继承自 TreeNode。每个节点可以有零到多个子节点，构成一棵树。

1
2
3
4
5
6
SELECT name, age FROM users WHERE age > 18 AND city = 'Beijing'

Logical Plan 树:
  Project [name, age]
  └── Filter (age > 18 AND city = 'Beijing')
      └── Relation [users]

Rule 是对树的变换函数。每个 Rule 接收一棵树，返回一棵变换后的树。变换通过模式匹配实现——匹配到特定的节点模式时替换为优化后的节点。

1
2
3
4
5
6
7
8
// 常量折叠规则的伪代码
object ConstantFolding extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan.transformExpressionsUp {
    case Add(Literal(a: Int), Literal(b: Int)) => Literal(a + b)
    case And(Literal(true), other) => other
    case And(Literal(false), _) => Literal(false)
  }
}

优化器就是一组 Rule 的有序集合（Batch）。每个 Batch 包含若干 Rule，按顺序对计划树反复应用，直到树不再变化（达到不动点）或达到最大迭代次数。

阶段一：Parsing

Spark SQL 使用 ANTLR4 生成的解析器把 SQL 文本解析成 Unresolved Logical Plan。这个阶段纯粹是语法分析，不涉及任何语义检查。

1
2
3
4
5
6
输入: SELECT name FROM users WHERE age > 18

输出 (Unresolved Logical Plan):
  Project [UnresolvedAttribute("name")]
  └── Filter (UnresolvedAttribute("age") > 18)
      └── UnresolvedRelation("users")

此时列名 “name” 和表名 “users” 都还是字符串，没有绑定到实际的表定义和数据类型。如果 SQL 有语法错误（比如 SELEC 拼错），这个阶段就会报 ParseException。

阶段二：Analysis

Analyzer 通过查询 Catalog（元数据目录）把 Unresolved Plan 中的引用绑定到实际的表和列。

1
2
3
4
5
6
7
8
9
10
11
12
输入 (Unresolved):
  Project [UnresolvedAttribute("name")]
  └── Filter (UnresolvedAttribute("age") > 18)
      └── UnresolvedRelation("users")

查询 Catalog:
  users 表 → schema: {name: STRING, age: INT, city: STRING}

输出 (Resolved):
  Project [name#12: STRING]
  └── Filter (age#13: INT > 18)
      └── Relation [users] {name#12, age#13, city#14}

每个列被分配一个唯一的 ID（如 name#12），避免同名列的歧义。如果引用了不存在的表或列，这个阶段会报 AnalysisException。

Analysis 阶段还会做隐式类型转换。比如 WHERE age > ‘18’ 中 age 是 INT 而 ‘18’ 是 STRING，Analyzer 会插入一个 Cast 节点把 ‘18’ 转为 INT。

阶段三：Optimization

Optimizer 是 Catalyst 的核心，包含数十个优化规则。主要的优化类别：

谓词下推（Predicate Pushdown）：把 Filter 节点尽量推到数据源附近，提前过滤掉不需要的行。

1
2
3
4
5
6
优化前:                          优化后:
  Project [name]                   Project [name]
  └── Filter (age > 18)           └── Join
      └── Join                        ├── Filter (age > 18)
          ├── Scan [users]            │   └── Scan [users]
          └── Scan [orders]           └── Scan [orders]

列裁剪（Column Pruning）：只读取查询实际需要的列，不读取多余的列。对于列式存储（Parquet/ORC），这可以减少大量 I/O。

常量折叠（Constant Folding）：在编译期计算常量表达式。WHERE year > 2020 + 5 变为 WHERE year > 2025。

Join 重排序（Join Reordering）：当多表 Join 时，选择代价最小的 Join 顺序。小表优先 Join 可以减少中间结果的大小。

简化表达式（Simplify Expressions）：WHERE NOT (age <= 18) 简化为 WHERE age > 18。

阶段四：Physical Planning

Physical Planner 把优化后的逻辑计划转化为可执行的物理计划。核心决策是为每个逻辑算子选择一个物理实现。

Join 策略选择是最典型的例子。同一个逻辑 Join，Spark 可以用不同的物理算法：

1
2
3
4
5
6
7
8
9
10
11
BroadcastHashJoin    小表广播到所有节点，大表端做 hash 匹配
                     条件: 小表 < spark.sql.autoBroadcastJoinThreshold (默认 10MB)

SortMergeJoin        两表按 Join key 排序后归并
                     条件: 两表都较大，默认策略

ShuffledHashJoin     按 Join key Shuffle 后在每个分区做 hash join
                     条件: 一侧表较小但超过广播阈值

BroadcastNestedLoop  广播小表，逐行匹配（用于非等值 Join）
                     条件: 非等值 Join 条件（如 a.x > b.y）

Physical Planner 根据表的大小统计信息（来自 Catalog 或运行时统计）选择最优策略。如果统计信息不准确，选择可能不是最优的——这也是 AQE（第 10 篇）要解决的问题。

实验：用 explain 观察四阶段变换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
val spark = SparkSession.builder.getOrCreate()
import spark.implicits._

// 准备测试数据
val users = Seq((1, "Alice", 25), (2, "Bob", 17), (3, "Carol", 30))
  .toDF("id", "name", "age")
users.createOrReplaceTempView("users")

// 查看完整的四阶段计划
spark.sql("SELECT name FROM users WHERE age > 18 AND 1 = 1")
  .queryExecution.toString
// 输出包含：
// == Parsed Logical Plan ==    (Unresolved)
// == Analyzed Logical Plan ==  (Resolved)
// == Optimized Logical Plan == (常量折叠把 1=1 消除了)
// == Physical Plan ==          (物理执行计划)

// 或者用 explain(true) 查看简化版
spark.sql("SELECT name FROM users WHERE age > 18 AND 1 = 1")
  .explain(true)

在输出中可以看到：Parsed Plan 中 1 = 1 还存在；Optimized Plan 中 1 = 1 被常量折叠规则消除了（AND true 等价于直接去掉）；Physical Plan 中显示了实际使用的 Scan 和 Filter 物理算子。

模式提炼

Catalyst 的 Tree + Rule 架构体现了编译器前端的经典设计：把变换分解为一组独立的规则，每个规则只负责一种模式匹配和替换，通过反复应用达到全局最优。

这个架构的优势是可扩展性：增加一种新的优化只需要添加一个 Rule，不需要修改现有代码。Spark 的扩展点 SparkSessionExtensions 允许第三方库注入自定义的 Analyzer Rule 和 Optimizer Rule。

同样的架构出现在编译器的 pass 管理器（LLVM 的 PassManager）、数据库的查询改写器（PostgreSQL 的 rewrite rules）、甚至 CSS 的层叠规则（多条规则按优先级依次应用到 DOM 节点）。

工程迁移表

Spark/Catalyst 概念 传统数据库对应 Flink 对应 编译器对应
Parsing（SQL → AST） SQL Parser Flink SQL Parser（Calcite） Lexer + Parser
Analysis（绑定名称和类型） Semantic Analysis Calcite Validator 语义分析（符号表查询）
Optimization（规则重写） Query Optimizer（RBO + CBO） Calcite Optimizer 优化 Pass（常量传播、死代码消除）
Physical Planning（策略选择） Plan Generator（执行计划生成） Flink Planner 指令选择（ISel）
Catalog 系统目录（pg_catalog） Flink Catalog 符号表
Rule Rewrite Rule Calcite RelOptRule Optimization Pass

常见误解

误解一：“DataFrame API 比 SQL 快”。DataFrame API 和 SQL 在 Analysis 之后进入完全相同的优化和执行路径。性能没有差异。区别只在于书写方式和编译期类型检查。

误解二：“Catalyst 只做基于规则的优化”。Catalyst 同时支持 RBO（Rule-Based Optimization）和 CBO（Cost-Based Optimization）。从 Spark 2.2 起，Join 重排序和 Aggregate 策略选择可以基于表的统计信息（行数、列的不同值数量等）做代价估算。

误解三：“explain 显示的计划就是实际执行的计划”。AQE 开启后（Spark 3.0+ 默认开启），实际执行计划可能在运行时被修改。explain 显示的是编译期的计划，运行时可能因为 Shuffle 后的统计信息变化而切换 Join 策略或合并分区。

练习

  1. 构造一条包含子查询的 SQL（如 WHERE id IN (SELECT id FROM orders)），用 explain(true) 观察 Optimizer 是否把子查询改写成了 Join。

  2. 对两个表做 Join，分别设置 spark.sql.autoBroadcastJoinThreshold 为 10MB 和 -1（禁用广播），用 explain 观察物理计划中 Join 策略的变化。

  3. 用 spark.sql(“ANALYZE TABLE users COMPUTE STATISTICS FOR ALL COLUMNS”) 收集统计信息后，再次 explain 一条涉及该表的查询。对比收集统计信息前后的物理计划是否变化。

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 已完成
07 容错机制：Lineage、Checkpoint 与推测执行 已完成
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树 本篇
09 DataFrame 与 Dataset：类型安全与性能的折中 下一篇
10 Adaptive Query Execution：运行时改写执行计划
11 数据源 API 与谓词下推：让存储层帮忙过滤
12 Structured Streaming：把流当成无界表
13 水位线与状态管理：流处理的时间语义
14 Spark on YARN / K8s：资源管理与部署模式
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料