4. Spark执行模型与作业生命周期

在 Databricks 上做性能优化，如果只盯着集群规格和参数，往往会错过真正的瓶颈来源。Spark 作业的执行过程本质上是一条从逻辑计划到物理执行、再到分布式任务调度的链路：Driver 把用户代码翻译成 DAG，DAGScheduler 依据宽依赖把 DAG 切成 Stage，TaskScheduler 再把每个 Stage 拆成一批 Task 分发到 Executor 上运行。理解这条链路，才能判断一次作业慢在“切分不合理”“Shuffle 太重”“内存不够”还是“数据倾斜”。

DAG 调度：从 RDD 血缘到 Stage 边界

Spark 的 DAG 由 RDD 之间的依赖关系构成。依赖分两类：窄依赖（narrow dependency）指父分区最多被子分区的一个分区使用，例如 map、filter、union；宽依赖（wide dependency）指父分区可能被多个子分区使用，典型是 groupByKey、reduceByKey、join（未做 map 端合并时）。DAGScheduler 从最终 RDD 反向遍历血缘，遇到宽依赖就切开，形成一个 Stage 边界。因此一个 Job 通常包含若干 Stage，Stage 之间通过 Shuffle 连接。

Stage 分两类：ShuffleMapStage 负责产生 Shuffle 数据并写入磁盘，ResultStage 负责产出最终结果。Stage 内部的所有窄依赖转换会被合并成一条流水线，在同一个 Task 中顺序执行，避免中间落盘。这也是为什么把多个窄依赖操作写在一起通常比反复触发 action 更高效。

Stage 与 Task 划分：并行度从哪来

Stage 确定后，Task 数量由该 Stage 最后一个 RDD 的分区数决定。对于 ShuffleMapStage，Task 数等于上游分区数；对于 ResultStage，Task 数等于最终 RDD 的分区数。分区数直接决定并行度，也决定每个 Task 处理的数据量。分区太少会导致单 Task 数据量过大、GC 压力高、Executor 利用率低；分区太多则 Task 调度开销和 Shuffle 小文件问题会凸显。

在 Databricks 中，AQE（Adaptive Query Execution）可以在运行时合并过小的 Shuffle 分区、拆分倾斜分区、把 SortMergeJoin 转成 BroadcastHashJoin。开启 AQE 后，Stage 边界和 Task 数量可能在运行时被动态调整，因此静态看 DAG 图并不完全等于实际执行计划。排查时应以 Spark UI 的 SQL/DataFrame 页签和 Stage 页签为准。

Shuffle 机制：性能分水岭

Shuffle 是跨 Stage 的数据重分布过程，也是绝大多数作业的性能分水岭。以 groupByKey 为例，上游 Task 按分区器把记录写入本地 Shuffle 文件，下游 Task 通过网络拉取属于自己的分区数据。这个过程涉及序列化、磁盘 I/O、网络传输和反序列化，成本远高于窄依赖。

Shuffle 的关键参数包括：spark.sql.shuffle.partitions（默认 200，决定 Shuffle 后分区数）、spark.shuffle.file.buffer（写缓冲）、spark.reducer.maxSizeInFlight（拉取缓冲）、spark.shuffle.compress（是否压缩）。在 Databricks 上，Shuffle 数据默认写入本地磁盘，若开启 Photon 或使用 Delta 优化，部分场景会走更高效的路径。

减少 Shuffle 的常见手段：用 reduceByKey 替代 groupByKey（map 端预聚合）、用 broadcast join 替代大表 join、用 coalesce 替代 repartition（避免不必要的全量 Shuffle）、在 ETL 中提前按分区键聚合并写入 Delta 表。

内存管理：Execution 与 Storage 的博弈

Executor 内存分为三块：Execution Memory（用于 Shuffle、Join、Sort、Aggregation 的临时缓冲）、Storage Memory（用于缓存 RDD/DataFrame 和广播变量）、User Memory（用户数据结构）。Execution 和 Storage 共享一块统一内存区域，可以互相借用，但 Execution 优先级更高，当 Execution 需要空间时会驱逐 Storage 中的缓存块。

常见内存问题包括：java.lang.OutOfMemoryError: GC overhead limit exceeded（GC 频繁但回收效果差）、Container killed by YARN for exceeding memory limits（堆外内存或整体超限）、以及 Executor 频繁 Full GC 导致 Task 超时。调优方向：提高 spark.executor.memory、调整 spark.memory.fraction（默认 0.6）、减少单 Task 数据量（增加分区数）、避免在 Driver 端 collect 大结果集。

瓶颈定位：从 Spark UI 读信号

定位瓶颈的核心工具是 Spark UI。重点看四个地方：

• Stage 页签：看每个 Stage 的 Duration、Task 数、Shuffle Read/Write 大小、Spill 情况。如果某个 Stage 的 Shuffle Write 远大于 Read，说明上游产生了大量中间数据。
• Task 分布：看 Task 的 Duration 分布。如果少数 Task 耗时远高于中位数（如 P99 是中位数的 5 倍以上），基本可以判定数据倾斜。
• SQL 页签：看物理计划中的 Exchange 节点（Shuffle）、Sort、Aggregate，以及 AQE 是否生效。
• Executor 页签：看 GC Time 占比。如果 GC Time 超过 Task Time 的 10%，说明内存压力大。

实际排查时，建议按“先看现象、再定类型、后调参数”的顺序推进。例如一个典型的 Databricks 作业：读取 Delta 表后做 groupBy 聚合再 join 维表。如果 Spark UI 显示某个 Stage 的 Shuffle Write 达到 200GB，且 Task Duration 的 P99 是中位数的 8 倍，那么瓶颈大概率是数据倾斜加 Shuffle 分区不合理。此时可以先开启 AQE 的倾斜 join 处理，再把 spark.sql.shuffle.partitions 从 200 调到 2000 左右，观察 Shuffle Read 是否更均匀、Task 耗时是否收敛。如果 GC Time 仍然很高，再考虑提高 Executor 内存或减少单 Task 处理的数据量。

总结来看，Spark 作业生命周期可以概括为：逻辑计划 → DAG → Stage → Task → Shuffle → 内存分配 → 执行与回收。性能优化的本质，是让每个环节的数据量和并行度匹配集群资源。DAG 调度决定切分是否合理，Stage/Task 划分决定并行度，Shuffle 决定跨节点成本，内存管理决定单 Task 能否稳定运行，而 Spark UI 提供的信号则是判断瓶颈在哪一环的依据。掌握这条链路，才能在 Databricks 上从“凭感觉调参”转向“按证据优化”。