4. Spark执行模型与任务调度

在 Databricks 上做大规模任务优化，第一步不是调参数，而是看懂 Spark 到底把你的代码拆成了什么。很多“任务跑得慢”“Executor 频繁 OOM”“集群加了几倍还是不快”的问题，根源都在执行模型这一层：DAG 怎么切 Stage、Stage 里有多少 Task、Shuffle 写了多少数据、Executor 内存被谁占满、并行度是否匹配数据量。理解这条链路，才能判断该改代码、改分区还是改集群。

DAG 与 Stage：从一行代码到可调度的执行单元

Spark 采用惰性求值。当你写下 df.filter(...).groupBy(...).agg(...).write(...) 时，Spark 并不会立刻执行，而是先构建一张有向无环图（DAG），把每个 RDD/DataFrame 转换记录为节点。真正触发执行的是 action，例如 count()、collect()、write()。此时 DAGScheduler 会从最终节点向前回溯，遇到宽依赖（需要跨分区重分布数据）就切出一个新的 Stage。

窄依赖（如 map、filter、select）可以在同一分区内流水线执行，不产生 Shuffle，因此被合并进同一个 Stage；宽依赖（如 groupBy、join、distinct、repartition）必须等待上游所有分区数据就绪，于是形成 Stage 边界。一个 Job 通常由多个 Stage 组成，Stage 之间按依赖顺序串行或并行执行。Stage 数量本身不是问题，但每个 Shuffle 边界都意味着落盘、网络传输和等待，这是大规模任务最主要的成本来源。

Task、分区与并行度：决定“同时干多少活”

Stage 内部会被拆成一组 Task，Task 数量等于该 Stage 输入的分区数（对 Shuffle 后的 Stage 则是 Shuffle 分区数）。每个 Task 处理一个分区，运行在一个 Executor 的 CPU core 上。因此并行度 ≈ 可用 core 数，而 Task 总数由分区数决定。二者关系决定了执行形态：

• 分区数远小于 core 数：大量 core 空闲，任务“吃不饱”，单 Task 数据量过大，容易 OOM 且拖长尾。
• 分区数远大于 core 数：core 利用率高，但调度开销、小文件、任务启动成本上升，Shuffle 元数据压力增大。
• 分区数 ≈ core 数的 2～4 倍：通常能兼顾利用率与调度开销，是较稳妥的起点。

并行度还受 spark.sql.shuffle.partitions（默认 200）控制。对 TB 级数据，200 个 Shuffle 分区往往严重不足，单个 Task 要处理数 GB 数据；对几十 MB 的小表，200 又过多，产生大量空 Task。合理做法是按 Shuffle 数据量估算：假设 Shuffle 写出 200 GB，希望每个分区 128～256 MB，则分区数约为 800～1600。Databricks 的 Adaptive Query Execution（AQE）可以在运行时合并小分区、拆分倾斜分区，但前提是开启并给足初始分区数。

Shuffle：大规模任务真正的瓶颈

Shuffle 是 Stage 之间的数据重分布过程，分为 map 端写和 reduce 端读。map 端把数据按 key 分区写入本地磁盘（Shuffle write），reduce 端通过网络拉取对应分区（Shuffle read）。它带来三类代价：磁盘 I/O、网络传输、以及为缓冲和聚合占用的内存。Shuffle 数据量越大，Stage 边界等待越久，失败重算成本也越高。

减少 Shuffle 的常见手段包括：用 broadcast join 把小表广播出去避免 join shuffle；用 map-side 预聚合降低写出量；提前 filter 和 select 减少参与 Shuffle 的列与行；对倾斜 key 加盐打散。判断是否需要优化，可以看 Spark UI 的 Shuffle Write/Read 大小与 Spill 指标：若 Spill（磁盘溢写）持续增长，说明单分区数据超出内存缓冲，必须调整分区数或内存配置。

Executor 内存模型：为什么加了内存还是 OOM

一个 Executor 的内存被划分为几块：预留内存（约 300 MB，固定开销）、用户内存（UDF、自定义对象）、执行内存（Shuffle、Join、Sort、Aggregation 的缓冲）和存储内存（缓存 RDD/DataFrame、广播变量）。执行内存与存储内存共享一个统一区域，可以互相借用，但执行内存优先级更高，必要时会把缓存数据逐出。

OOM 常见于三种情况：一是单分区数据过大，聚合或排序时执行内存装不下；二是 UDF 或 collect() 把大量数据拉到 Driver 或 Executor 堆内；三是缓存过多 DataFrame 挤占执行内存。诊断时先看 Spark UI 的 Storage 页和 Executor 页，确认是执行内存溢出还是存储占用过高。对应手段：提高 Shuffle 分区数降低单分区数据量、避免在 Executor 端 collect 大结果、用 MEMORY_AND_DISK 替代纯内存缓存、必要时增大 spark.executor.memory 并同步调整 spark.memory.fraction。

把执行模型落到调优动作上

面对一个慢任务，可以按以下顺序定位：先看 Spark UI 的 Stage 列表，找出耗时最长和 Shuffle 读写最大的 Stage；再看该 Stage 的 Task 分布，若存在明显长尾（个别 Task 远慢于中位数），说明数据倾斜，需要加盐或 AQE 倾斜处理；若所有 Task 都慢且 Spill 很大，说明分区数不足或内存偏小；若 Task 数远大于 core 数且每个 Task 都很短，说明分区过碎，应合并分区或减少 spark.sql.shuffle.partitions。

下面是一份可直接对照使用的参数检查清单，适用于 Databricks 集群上的大规模 Shuffle 任务：

最后要建立的一个判断习惯是：并行度不是越大越好，而是要与数据量、单分区处理成本和集群 core 数匹配。Stage 边界是 Shuffle 成本的来源，Task 数是并行度的上限，Executor 内存决定单 Task 能处理多大数据。把这三者与 Spark UI 的实际指标对齐，才能把“加机器”变成“改对地方”。