Spark 执行模型源码笔记：从懒执行到 QueryExecution

最近开始读 Spark 源码，第一件事就是回答自己 Day 1 笔记里的疑问：DataFrame 为什么是懒执行的？“懒”到底懒在哪一行代码上？

概念层面很好讲：Transformation 只是构建计划，Action 才触发计算。但源码里“触发”是怎么传导的？我翻了一遍 QueryExecution.scala，发现答案比想象中干净 —— 整个执行链就是一连串 LazyTry 串起来的。

01先复习：DataFrame 是声明式 API

用 Spark 写数据处理，你描述的是“要什么”，而不是“怎么算”：

python

b = (
    orders
    .filter("order_status = 'delivered'")
    .select("order_id", "customer_id")
)

这一步只是往逻辑计划树上挂节点，什么都没算。真正的计算要等到 Action（show() / count() / write() / collect()）出现。

python

b.show()      # 第一次触发执行
n = b.count() # 又一次 Action，会再次触发计算！
b.write(...)  # 同样是 Action

除非 b.cache()，否则 Spark 不会自动复用上一次的结果，每个 Action 都会重算。这是懒执行最常见的坑。

02一份查询要经过五层计划

从用户代码到真正执行，计划被改写了五次：

text

DataFrame API / SQL
        ↓
Parsed Logical Plan       ← 语法解析后的"源码"
        ↓
Analyzed Logical Plan     ← 绑定元数据（列名、表、类型）
        ↓
Optimized Logical Plan    ← 优化器改写（谓词下推、常量折叠…）
        ↓
Physical Plan            ← 选定具体算子与执行策略
        ↓
Executed Plan            ← 准备就绪、可执行的计划

我的 Day 1 笔记只到这里。但“五层”是抽象描述，源码里它们是怎么落地的？这是这次读源码的重点。

03源码入口：QueryExecution.scala

QueryExecution 是执行模型的枢纽，负责把 LogicalPlan 一步步变成可执行的 SparkPlan。文件在：

text

sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala

它内部几乎每个阶段都是一个 LazyTry 包装的懒值，从优化的角度看这非常“教科书”：

scala

// 阶段 1：逻辑优化
private val lazyOptimizedPlan = LazyTry {
  assertCommandExecuted()
  executePhase(QueryPlanningTracker.OPTIMIZATION) {
    val plan =
      sparkSession.sessionState.optimizer.executeAndTrack(withCachedData.clone(), tracker)
    plan.setAnalyzed()
    plan
  }
}

// 阶段 2：从逻辑计划生成物理计划
private val lazySparkPlan = LazyTry {
  assertOptimized()
  val plan = executePhase(QueryPlanningTracker.PLANNING) {
    QueryExecution.createSparkPlan(planner, optimizedPlan.clone())
  }
  attachTransaction(plan)
}

// 阶段 3：准备执行（加各种 preparation 规则）
private val lazyExecutedPlan = LazyTry {
  assertOptimized()
  val plan = executePhase(QueryPlanningTracker.PLANNING) {
    QueryExecution.prepareForExecution(preparations, sparkPlan.clone())
  }
  tracker.setReadyForExecution()
  plan
}

// 阶段 4：真正产出 RDD
val lazyToRdd = LazyTry {
  new SQLExecutionRDD(executedPlan.execute(), sparkSession.sessionState.conf)
}

这就是“懒”的代码级答案：toRdd 才是被 Action 直接触碰的入口，而它依赖 executedPlan，executedPlan 依赖 sparkPlan，sparkPlan 依赖 optimizedPlan —— 层层 lazy，谁先被访问谁才初始化。

读这份代码还有三个细节值得记：

每一步都在 clone()。注释写得很直白：clone the plan to avoid sharing the plan instance between different stages like analyzing, optimizing and planning。计划树是共享结构，分析、优化、规划阶段要改它，不克隆就会互相污染。
每个阶段都先 assert 上一阶段。assertOptimized() 出现在 sparkPlan 里、assertOptimized() 又出现在 executedPlan 里。这是一种显式的依赖契约：宁可报错，也不允许跳过阶段直接取计划。
executePhase 记录耗时。QueryPlanningTracker.OPTIMIZATION / PLANNING 这些常量把每个阶段的时间分开统计，也就是你在 Spark UI 里看到的各阶段耗时来源。

04LazyTry：为什么 Spark 5.0 不直接用 lazy val

看到 LazyTry 这个名字，我第一反应是“这不就是个 lazy val 吗”。读了 core/src/main/scala/org/apache/spark/util/LazyTry.scala 才发现它解决了 Scala lazy val 的两个问题：

scala

private[spark] class LazyTry[T](initialize: => T) extends Serializable {
  private lazy val tryT: Try[T] = Utils.doTryWithCallerStacktrace { initialize }

  @volatile private var materialized: Boolean = false

  def get: T = {
    val isFirstAccess = !materialized
    materialized = true
    Utils.getTryWithCallerStacktrace(tryT, isFirstAccess)
  }
}

失败不重试。Scala 的 lazy val 初始化抛异常后，字段保持未初始化状态，下次访问会再算一遍——既浪费性能，还可能重复副作用。LazyTry 把结果存进 Try，失败就记住失败，下次直接 rethrow。
锁的粒度。Scala lazy val 初始化时要锁住外层对象，多线程场景可能跟别的锁互相等成死锁。LazyTry 只锁它自己这个 wrapper，锁的范围小得多。

对 QueryExecution 这种会被并发访问、失败就该立即暴露的执行计划来说，这两个特性都很关键。

05优化器入口：RuleExecutor 的规则批

optimizer.executeAndTrack(...) 最终走到 RuleExecutor.execute。优化器不是“一整套优化”，而是一批一批的规则串行执行：

scala

def execute(plan: TreeType): TreeType = {
  var curPlan = plan
  // ...
  batches.foreach { batch =>
    var iteration = 1
    var lastPlan = curPlan
    // batch 内的规则串行跑，直到 plan 不再变化（fixedPoint）或跑完一遍（once）
  }
}

每个 batch 有自己的策略：Once 跑一遍，FixedPoint 一直跑到结果不再变化。比如谓词下推、列裁剪这类规则会组成 batch，按顺序应用。这也是为什么 EXPLAIN 里的优化后计划总是“看起来被改了很多”——它是几十条规则迭代的产物。

06Partition：并行计算的基本单位

说完执行链，回到数据层面。Partition 是 Spark 把数据切成的分片，也是 Task 的输入单位：

text

Partition = 数据分片
Task      = 处理一个分片的任务
Executor  = 运行 Task 的进程

关键认知：文件是存储层概念，Partition 是计算层概念。一个大文件可能被切成多个 Partition，多个小文件也可能合并进一个 Partition。读文件时会自动分区：

python

orders = spark.read.csv("orders.csv")  # 不写 repartition 也会自动切

手动调整分区用两个方法，语义差别很大：

方法	行为	是否触发 Shuffle
`df.repartition(8)`	增加/重排分区	✅ 会
`df.coalesce(2)`	合并分区（通常只减不增）	❌ 一般不触发

默认值也分场景：

场景	默认	配置项
读文件	每个输入分区最大约 128 MB	`spark.sql.files.maxPartitionBytes`
Shuffle 之后	200 个分区	`spark.sql.shuffle.partitions`

注意 128 MB 是最大目标切分大小，不是每 128 MB 一定一个分区，小文件会合并、大文件会切分。

07RDD 三层抽象：下一篇的引子

QueryExecution.toRdd 最终产出一个 RDD[InternalRow]。一个 RDD 只需要回答三个问题：

text

A list of partitions                # 我有多少个分区
A function for computing each split # 每个分区怎么算
A list of dependencies              # 我依赖哪些上游 RDD

第三个问题（依赖）直接引出下一篇的主题：宽依赖 vs 窄依赖，以及 Shuffle 为什么是 Stage 的边界。这也是 Day 1 笔记里标注“待续”的部分。

08小结

这次读源码，把“懒执行”从一个概念还原成了代码事实：

DataFrame 的 Transformation 只构建逻辑计划树，Action 才触碰 toRdd。
QueryExecution 用一条 LazyTry 链把优化、规划、执行串起来，层层 lazy、层层 clone、层层 assert。
LazyTry 比 Scala lazy val 更适合作执行计划：失败不重试、锁粒度小。
优化器是几十条规则的批处理，不是一坨“魔法”。
Partition 是并行单位，128 MB 是最大目标切分，coalesce 通常不触发 Shuffle。

下一篇：宽依赖、窄依赖与 Stage 划分，继续沿着 RDD.dependencies 往下读。

Spark 执行模型源码笔记：从懒执行到 QueryExecution

01先复习：DataFrame 是声明式 API

02一份查询要经过五层计划

03源码入口：QueryExecution.scala

04LazyTry：为什么 Spark 5.0 不直接用 lazy val

05优化器入口：RuleExecutor 的规则批

06Partition：并行计算的基本单位

07RDD 三层抽象：下一篇的引子

08小结

相关文章

Spark 执行模型笔记（二）：宽窄依赖、Shuffle 与 Stage

湖仓一体不是一次切换：旁路双写、直连湖与读路径的回滚设计

Kafka 可靠性语义笔记：从分区到 Exactly-Once