【极速学习spark源码】一条 SQL 在 Spark 中的一生

基于 Spark 4.2，分支 branch-4.2
阅读时长约 10分钟 · 入门到中级

引子

写下这一行, 你已经写过几百次的代码：

spark.sql("SELECT name FROM users WHERE age > 18").show()

这句话背后，Spark 究竟做了哪些事？

如果你只能说出"解析、优化、生成 RDD"，那么这篇文章想带你逐行走完这趟旅程——看清每个阶段的入口在哪里、产物是什么、什么时候发生、为什么这样设计。

读完之后，下次再看到一份执行计划，你应该能反过来推断它经过了哪些规则。

一、流程图：九个阶段，一根流水线

用户代码   │   ▼┌──────────────────── Driver 端，单线程 ────────────────────┐│                                                            ││  1. SparkSession.sql        ← 入口                         ││  2. Parser                  字符串 → Unresolved LogicalPlan││  3. Dataset.ofRows          创建 QueryExecution            ││  4. Analyzer                绑定 schema / 解析名字          ││  5. Command 急执行          （只 DDL/DML 走这步）            ││  6. Cache 替换 + 规范化     命中 cache 子树替换               ││  7. Optimizer               等价变换，求最优逻辑计划          ││  8. Planner                 LogicalPlan → SparkPlan        ││  9. Prepare for Execution   插 Exchange/Sort/Codegen 等    ││                                                            │└────────────────────────────────────────────────────────────┘   │   ▼ executedPlan.execute()┌──────────────────── 分布式执行 ──────────────────────────┐│  RDD → DAGScheduler → TaskScheduler → Executor → Task    │└──────────────────────────────────────────────────────────┘

两个最容易踩坑的认知：

前 8 步全部在 Driver 单线程跑。SQL "慢"如果慢在编译期，加 executor 数量不会有任何帮助。
DataFrame 并非完全 lazy。spark.sql(...) 返回的瞬间 Analyzer 已经跑完（要知道 schema 才能给你返回 DataFrame）；Optimizer / Planner 才真正等到 action 触发。

二、入口：SparkSession.sql

打开 SparkSession.scala:563：

def sql(sqlText: String, args: Array[_]): DataFrame = {  sql(sqlText, args, new QueryPlanningTracker)}

每次调用都新建一个 QueryPlanningTracker。这个小对象贯穿整条路径，每个阶段都用它打卡。后面你能看到为什么这件事很重要。

真正干活的重载在 SparkSession.scala:528：

private[sql] def sql(sqlText: String, args: Array[_], tracker: QueryPlanningTracker)    : DataFrame = withActive {  val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {    val parsed = sessionState.sqlParser.parsePlanWithParameters(sqlText, paramContext)    // ... 参数绑定处理 ...    parsed  }  Dataset.ofRows(self, plan, tracker)}

到这里就两件事：

measurePhase("parsing") { ... } 包住 Parser，记录耗时
把解析出来的 LogicalPlan 交给 Dataset.ofRows

干净利落。

三、Parser：字符串变成树

Parser 的本职是 ANTLR4 工作。入口在 SparkSqlParser.scala:85，再往下就是 parsers.scala:59：

protected def parse[T](command: String)(toResult: SqlBaseParser => T): T = {  val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))  // ...  val tokenStream = new CommonTokenStream(lexer)  val parser = new SqlBaseParser(tokenStream)  // ...}

注意 UpperCaseCharStream——这就是 Spark SQL 关键字大小写不敏感的实现方式：lexer 看到的字符流统一是大写，但原始 token 文本保留。

从 AST 到 LogicalPlan 走访问者模式，由 SparkSqlAstBuilder（继承自 AstBuilder）完成。

代码:：

val parsed = spark.sessionState.sqlParser.parsePlan(  "SELECT name FROM users WHERE age > 18")println(parsed)

输出：

'Project ['name]
+- 'Filter ('age > 18)
   +- 'UnresolvedRelation [users], [], false

注意：表不存在不会在这里报错。所有以 Unresolved 开头的节点都意味着"还没问过 catalog"。这是 Parser 的边界。

四、QueryExecution：阶段流水线的容器

接力棒在 SparkSession.scala:559 传给 Dataset.ofRows，看 Dataset.scala:111：

def ofRows(sparkSession: SparkSession, logicalPlan: LogicalPlan): DataFrame =  sparkSession.withActive {    val qe = sparkSession.sessionState.executePlan(logicalPlan)    if (!qe.isLazyAnalysis) qe.assertAnalyzed()    new Dataset[Row](qe, () => RowEncoder.encoderFor(qe.analyzed.schema)) }

两件事：

创建一个 QueryExecution，把未解析的逻辑计划塞进去
立即调用 assertAnalyzed() ——也就是上面说的"DataFrame 创建时 Analyzer 已经跑了"

为什么必须现在跑？因为接下来 RowEncoder.encoderFor(qe.analyzed.schema)（Dataset.scala:115）需要 schema，而 schema 只有在 Analyze 之后才有。

QueryExecution 本身是什么？它是后续所有阶段的懒加载容器。打开 QueryExecution.scala，看行192-394 那一长串你会发现一个清晰的模式：

private val lazyAnalyzed       = LazyTry { ... }   // 192private val lazyCommandExecuted = LazyTry { ... }  // 215private val lazyNormalized     = LazyTry { ... }   // 280private val lazyWithCachedData = LazyTry { ... }   // 289private val lazyOptimizedPlan  = LazyTry { ... }   // 313private val lazySparkPlan      = LazyTry { ... }   // 337private val lazyExecutedPlan   = LazyTry { ... }   // 355val      lazyToRdd             = LazyTry { ... }   // 378

每一阶段对应一个 LazyTry，外面包一层 def：

def analyzed:      LogicalPlan = lazyAnalyzed.getdef optimizedPlan: LogicalPlan = lazyOptimizedPlan.getdef sparkPlan:     SparkPlan   = lazySparkPlan.getdef executedPlan:  SparkPlan   = lazyExecutedPlan.getdef toRdd:         RDD[InternalRow] = lazyToRdd.get

调试 SQL 编译过程的所有套路都建立在这上面：

df.queryExecution.analyzed       // 看 Analyzer 产物df.queryExecution.optimizedPlan  // 看 Optimizer 产物df.queryExecution.sparkPlan      // 看 Planner 产物df.queryExecution.executedPlan   // 看 Prepare 后的最终物理计划df.queryExecution.tracker.phases // 看每个阶段耗时

记住这五行。看任何 Spark SQL 性能问题，没有不是从这里入手的。

五、Analyzer：把名字绑定到东西

QueryExecution.scala:192：

private val lazyAnalyzed = LazyTry {  val plan = executePhase(QueryPlanningTracker.ANALYSIS) {    analyzer.executeAndCheck(sqlScriptExecuted, tracker)  }  tracker.setAnalyzed(plan)  plan}

具体到 Analyzer.scala:331：

def executeAndCheck(plan: LogicalPlan, tracker: QueryPlanningTracker): LogicalPlan = {  if (plan.analyzed) {    plan  } else {    def runAnalysis(): LogicalPlan =      HybridAnalyzer.fromLegacyAnalyzer(legacyAnalyzer = this, tracker = tracker).apply(plan)    // ...  }}

注意：4.x 引入了 HybridAnalyzer，老文章里直接讲的 Analyzer extends RuleExecutor 这条单链已经不准确。本文不展开 Hybrid 机制，只把它当成"Analyzer 的入口"。

Analyzer 的工作可以这样概括：一组规则按 fixed-point（不动点）反复跑，直到计划不再变化。每条规则负责消除一种 Unresolved：

Unresolved	谁来解析	解析成什么
`UnresolvedRelation`	`ResolveRelations`	去 catalog 查表，变成 `LocalRelation` / `HiveTableRelation` / `DataSourceV2Relation` 等
`UnresolvedAttribute`	`ResolveReferences`	列名变成有 `ExprId` 的 `AttributeReference`
`UnresolvedFunction`	`ResolveFunctions`	函数名变成具体的 `Expression` 实现

代码：

spark.range(100).toDF("id").createOrReplaceTempView("t")val df = spark.sql("SELECT id FROM t WHERE id > 10")println("=== Parsed ===")println(df.queryExecution.logical)   // 未解析println("=== Analyzed ===")println(df.queryExecution.analyzed)  // 已解析

输出：

=== Parsed ===
'Project ['id]
+- 'Filter ('id > 10)
   +- 'UnresolvedRelation [t], [], false

=== Analyzed ===
Project [id#1L]
+- Filter (id#1L > cast(10 as bigint))
   +- SubqueryAlias t
      +- View (`t`, [id#1L])
         +- Project [id#0L AS id#1L]
            +- Range (0, 100, step=1, splits=Some(1))

注意几个变化：

'id （未解析的 Attribute）变成 id#0L （已绑定 ExprId，类型是 long）
UnresolvedRelation 被替换成具体的 SubqueryAlias + Project + Range
10 被自动 cast(10 as bigint) ——类型隐式转换也在 Analyzer 完成

如果有报错（找不到表、列名歧义、类型不匹配），抛出的 AnalysisException 也来自这一步。

六、Optimizer：等价变换求最优

QueryExecution.scala:313：

private val lazyOptimizedPlan = LazyTry {  assertCommandExecuted()  executePhase(QueryPlanningTracker.OPTIMIZATION) {    val plan = sparkSession.sessionState.optimizer      .executeAndTrack(withCachedData.clone(), tracker)    plan.setAnalyzed()    plan  }}

Optimizer 也是 RuleExecutor，但规则更多更狠。它的工作精神可以一句话概括：结果集等价的前提下，找一棵开销最小的计划树。

最经典的几条：

规则	干什么
`ConstantFolding`	`1 + 1` → `2`、`1 = 1` → `true`
`BooleanSimplification`	`true AND x > 5` → `x > 5`
`PushDownPredicates`	把 `WHERE` 推到 scan 节点附近，减少数据量
`ColumnPruning`	只读取真正用到的列
`CollapseProject`	合并相邻的 `Project`
`LimitPushDown`	把 `LIMIT` 下推到 union / join 各分支

代码：

val df = spark.sql("""  SELECT id FROM t  WHERE 1 = 1 AND id > 5 AND id IS NOT NULL""")println("=== Analyzed ===")println(df.queryExecution.analyzed)println("=== Optimized ===")println(df.queryExecution.optimizedPlan)

输出：

=== Analyzed ===
Project [id#13]
+- Filter (((1 = 1) AND (id#13 > 5)) AND isnotnull(id#13))
   +- SubqueryAlias users
      +- View (`users`, [id#13, name#14, age#15])
         +- Project [_1#3 AS id#13, _2#4 AS name#14, _3#5 AS age#15]
            +- LocalRelation [_1#3, _2#4, _3#5]

=== Optimized ===
Project [_1#3 AS id#13]
+- Filter (_1#3 > 5)
   +- LocalRelation [_1#3, _2#4, _3#5]

1 = 1 被 ConstantFolding 折叠
true AND ... 被 BooleanSimplification 消去
id > 5 已经隐含了 id IS NOT NULL，多余的 IS NOT NULL 被 BooleanSimplification / InferFiltersFromConstraints 处理

关键提醒：Optimizer 的所有变换必须结果集等价。如果你看到 Optimizer 把你的 SQL 改成了一个"明显不一样"的形式但结果正确，不要慌——它就是干这个的。

七、Planner：从逻辑到物理

QueryExecution.scala:337：

private val lazySparkPlan = LazyTry {  assertOptimized()  val plan = executePhase(QueryPlanningTracker.PLANNING) {    QueryExecution.createSparkPlan(planner, optimizedPlan.clone())  }  attachTransaction(plan)}

createSparkPlan 的实现非常诚实，只有一行有效代码（QueryExecution.scala:808）：

def createSparkPlan(planner: SparkPlanner, plan: LogicalPlan): SparkPlan = {  // TODO: We use next(), i.e. take the first plan returned by the planner, here for now,  //       but we will implement to choose the best plan.  planner.plan(ReturnAnswer(plan)).next()}

源码里那条 TODO 揭示了一个真相：Spark 的 planner 理论上能产出多个候选物理计划（Iterator[SparkPlan]），但当前实现就是粗暴地 .next() 取第一个。所谓"基于代价的优化"（CBO）主要发生在 Optimizer 阶段，而不是这里。

Planner 的工作机制是 Strategy 模式。看 SparkPlanner.scala:38：

override def strategies: Seq[Strategy] =  experimentalMethods.extraStrategies ++    extraPlanningStrategies ++ (    LogicalQueryStageStrategy ::    PythonEvals ::    new DataSourceV2Strategy(session) ::    V2CommandStrategy ::    FileSourceStrategy ::    DataSourceStrategy ::    SpecialLimits ::    Aggregation ::    Window ::    WindowGroupLimit ::    JoinSelection ::         // ← BHJ / SMJ / SHJ 在这里决定    InMemoryScans ::    SparkScripts ::    Pipelines ::    BasicOperators ::    EventTimeWatermarkStrategy :: Nil)

每个 Strategy 都是一个偏函数：匹配某种 LogicalPlan 形态，吐出对应的 SparkPlan。比如：

FileSourceStrategy 把 LogicalRelation(HadoopFsRelation) 翻译成 FileSourceScanExec
JoinSelection 看 join 两边的 size 和 hint，决定生成 BroadcastHashJoinExec / SortMergeJoinExec / ShuffledHashJoinExec
Aggregation 选择 HashAggregateExec 还是 SortAggregateExec

Strategy 顺序很重要：靠前的 Strategy 优先匹配。所以 LogicalQueryStageStrategy（AQE 的）排在最前面。

八、Prepare for Execution：最后一公里

走到这里你已经有一棵 SparkPlan 了，但不能直接执行——它还缺：

shuffle / sort（如果上游没满足下游的分区或排序要求）
WholeStageCodegen 包装（合并能融合的算子）
AQE 包装（如果启用）
subquery 物化
……

这些都在 prepareForExecution 里加。看 QueryExecution.scala:752：

private[execution] def preparations(    sparkSession: SparkSession,    adaptiveExecutionRule: Option[InsertAdaptiveSparkPlan] = None,    subquery: Boolean): Seq[Rule[SparkPlan]] = {  adaptiveExecutionRule.toSeq ++  Seq(    CoalesceBucketsInJoin,    PlanDynamicPruningFilters(sparkSession),    PlanSubqueries(sparkSession),    RemoveRedundantProjects,    EnsureRequirements(),    InsertSortForLimitAndOffset,    ReplaceHashWithSortAgg,    RemoveRedundantSorts,    RemoveRedundantWindowGroupLimits,    DisableUnnecessaryBucketedScan,    ApplyColumnarRulesAndInsertTransitions(...),    CollapseCodegenStages()  ) ++ (if (subquery) Nil else Seq(ReuseExchangeAndSubquery))}

要记住的就两条：

EnsureRequirements 是 Exchange 的来源。所有的 shuffle 节点都不是 Planner 生成的，是这一步根据 requiredChildDistribution 推断出来的。
CollapseCodegenStages 是 WholeStageCodegen 的来源。看到 *(N) HashAggregate 这种带星号的算子，就是被它打包了。

关于 AQE 的提醒：如果 AQE 启用，InsertAdaptiveSparkPlan 会把整个计划包进 AdaptiveSparkPlanExec——一个 leaf node。后面所有规则因此都变成 no-op（leaf node 没有子树可改）。真正的 prepare 推迟到 AQE 内部按 stage 逐段做。

九、Execute：进入分布式世界

QueryExecution.scala:378：

val lazyToRdd = LazyTry {  new SQLExecutionRDD(executedPlan.execute(), sparkSession.sessionState.conf)}

executedPlan.execute() 触发整棵物理计划的递归 doExecute() 调用，从叶子（scan）到根，最终构建出一棵 RDD 血缘。

到这里 Catalyst 的工作就完成了。调用 df.show() / df.collect() / df.count() 等 action 时，才会触发 SparkContext.runJob(rdd, ...)，由 DAGScheduler 把 RDD 切成 Stage 和 Task 推到 Executor。

那是另一篇文章的故事（系列文章 #2）。

十、实战：把九个阶段一次看个够

把所有工具组合起来：

spark.range(100).toDF("id").createOrReplaceTempView("t")val sql = "SELECT id FROM t WHERE id > 10 AND 1 = 1"val df = spark.sql(sql)println("\n=== 1. Parser 单独跑 ===")println(spark.sessionState.sqlParser.parsePlan(sql))println("\n=== 2. Analyzed ===")println(df.queryExecution.analyzed)println("\n=== 3. Optimized ===")println(df.queryExecution.optimizedPlan)println("\n=== 4. SparkPlan (未 prepare) ===")println(df.queryExecution.sparkPlan)println("\n=== 5. ExecutedPlan (已 prepare) ===")println(df.queryExecution.executedPlan)// 触发执行，让 tracker 数据完整df.collect()println("\n=== 6. 各阶段耗时 ===")df.queryExecution.tracker.phases.foreach { case (name, summary) =>  println(f"$name%-20s ${summary.endTimeMs - summary.startTimeMs}%5d ms")}

把这段在你本地的 Spark 4.x 上跑一遍，对照前面九节的描述，你会看到：

第 1 步出 UnresolvedRelation、UnresolvedAttribute
第 2 步所有 Unresolved 消失，类型补齐
第 3 步 1 = 1 折叠掉
第 4 步出现具体物理算子（*Range、*Filter、*Project）
第 5 步可能多了 WholeStageCodegen 包装
第 6 步打印出 parsing / analysis / optimization / planning 四个阶段的耗时

十一、总结

回顾：

spark.sql(sqlText)      └─ SparkSqlParser.parsePlan          → Unresolved LogicalPlan            └─ Dataset.ofRows                  └─ QueryExecution (lazy 容器)                        ├─ analyzer.executeAndCheck    → Analyzed LogicalPlan                        ├─ cacheManager.useCachedData  → Normalized                        ├─ optimizer.executeAndTrack   → Optimized LogicalPlan                        ├─ planner.plan(...).next()    → SparkPlan                        ├─ prepareForExecution         → ExecutedPlan                        └─ executedPlan.execute()      → RDD[InternalRow]                                                              │                                                              ▼                                                   action → DAGScheduler → ...

三个最值得带走的认知：

Spark SQL 的"编译期"很长。前 8 步全部串行跑在 Driver 上。如果你的 SQL 已经在 df.queryExecution.executedPlan 卡了好几秒，加机器没用——需要看 Optimizer / Planner 是哪条规则慢。
DataFrame 的 lazy 不彻底。Analyzer 在 spark.sql(...) 返回前就跑完了。表不存在、列名错这类错误立刻抛；但 Optimizer/Planner 要等 action。
df.queryExecution.{analyzed, optimizedPlan, sparkPlan, executedPlan, tracker} 是性能调优的瑞士军刀。任何 Spark SQL 性能问题都从这里下手。

每天花费10分钟学习spark，让你技术之路走得更稳、更快。
喜欢的点个关注。