上一篇从 QueryExecution 里看到了“懒执行”的代码真相。但逻辑计划最终怎么变成一堆真正并行跑的任务？这就引出执行模型的下半场：依赖、Shuffle 和 Stage。

先给一句话全局：Spark 的任务执行，就是从“数据”出发，画一张依赖图，在图里切出若干段，每段内部并行算、段与段之间靠 Shuffle 衔接。

01三个实体与并行度

  • Partition（分区）：数据被切成的小块，并行计算的基本单位
  • Task（任务）：处理一个 Partition 的工作单元
  • Executor（执行器）：真正跑 Task 的进程

关系：一个 Executor 上可以跑多个 Task，每个 Task 负责一个 Partition。

并行度 = 有多少个 Partition。分区多，任务就多，能同时算的越多。

这个等式是后面所有判断的出发点。

02窄依赖 vs 宽依赖：全部的核心

Spark 把算子作用在很多 Partition 上，每个下游 Partition 要“看”几个上游 Partition 才能算出来，决定了两种依赖。

窄依赖（Narrow Dependency）：一个下游分区只看一个上游分区

text
上游:  P0   P1   P2
        |    |    |
下游:  P0'  P1'  P2'   ← 一一对应

filter、select、map 都是。每个分区分头算，互不干扰——大户人家各自扫门前雪，一个分区算完就完事，不需要等其他分区。

宽依赖（Wide Dependency）：一个下游分区要看多个上游分区

text
上游:  P0   P1   P2
        \    |    /
         \   |   /      ← 下游每个分区都要收上游所有分区里的同类数据
         v   v   v
        P0'  P1' ...

groupBy、join 都是。要把分散在不同分区里的“同一类”数据聚到一起——比如 delivered 状态的订单散落在 P0、P1、P2，要统计总量，就得把三个分区的 delivered 全捞到同一个下游分区。这就发生了数据的跨分区搬家。

03Shuffle = 宽依赖引发的“搬家”

Shuffle 就是数据因为需要重新分组，被跨分区（甚至跨机器）重新分发的过程。

凡是产生宽依赖的算子（groupBy、join、repartition 等）都会触发 Shuffle。它贵，是因为搬家不免费：

  • 上游分区把数据按 key 打散，写成一堆临时 Shuffle 文件
  • 下游分区去读上游各机器产生、属于自己的那些文件
  • 涉及网络传输、磁盘读写、序列化

一句话记：窄依赖“原地算”，宽依赖“要搬家”，搬家 = Shuffle，搬家最费钱。

补充一个概念细节：Shuffle 不是瞬间动作，而是一条“上游写文件 → 下游读文件”的流水线。写端（Map 侧）按 key 哈希把数据归桶、每个桶落成临时文件；读端（Reduce 侧）启动后去抓取属于自己分区的分片，合并、排序后继续算。上游 3 分区、下游 2 分区，老式实现会产生 3×2 = 6 个分片文件——这也是“小文件问题”的源头之一。

04Stage：被 Shuffle 切开的“段”

Stage 是一串中间没有 Shuffle 的计算步骤。Shuffle 是 Stage 之间的分界线。

为什么这么切？

  • 一个 Stage 内部全是窄依赖 → 各 Partition 互不依赖 → 可以并行、连续地算，不用等别人
  • 一旦碰到 Shuffle → 下游 Stage 必须等上游 Stage 所有数据都“搬完”才能开始

看一段链式操作：

python
orders.filter(...)                # 窄依赖
      .select(...)                # 窄依赖
      .groupBy(...).count()       # 宽依赖 → Shuffle！
      .orderBy(...)               # 又一次重排 → 又一次 Shuffle

切出来就是：

text
Stage 0: 读 → filter → select ──Shuffle──▶ Stage 1: groupBy+count ──Shuffle──▶ Stage 2: orderBy

记法：你看到多少个 Shuffle，基本就有“Shuffle 数 + 1”个 Stage。 Shuffle 就是切蛋糕的刀。

05Job / Stage / Task：三个层级

层级关系：

text
Application（应用）
   └── Job ① （一个 Action 触发一个 Job）
          ├── Stage 0
          ├── Stage 1   ← 被 Shuffle 切开
          └── Stage 2
                  └── Task 若干（= 该 Stage 的分区数）

Job 的触发单位是 Action（show() / count() / write() / collect()）。你调一次 action，就产生一个 Job——前面攒了多少 transformation，action 一来全部一口气执行，这就是一个 Job。

Stage 是 Job 内部被 Shuffle 切成的段，段内全是窄依赖。

Task 是真正被 Executor 执行的最小单元，一个 Partition 对应一个 Task。

核心区别一句话：

Job 是用户（Action）划的界；Stage 是系统（Shuffle）划的界。

在 Spark UI（driver 的 4040 端口）里看得最清楚：最上面是 Jobs 列表，每个 Action 一个；点进一个 Job 看到 Stage DAG 图，被箭头（Shuffle）分成几段；每段里是 Tasks，一堆并行的小方块。

记忆链：Action 数 = Job 数；Job 里 Stage 数 ≈ Shuffle 数 + 1；Stage 里 Task 数 = 分区数。

06repartition vs coalesce：两个“调分区”的工具

repartition(n)：重排成 n 个分区，因为要重新分配数据 → 产生 Shuffle。适用：增加并行度、或按 key 归拢数据（repartition(8, "customer_id")）。

coalesce(n)：把相邻分区拼一起 → 尽量不产生 Shuffle。适用：减少分区、少输出文件。

关键坑：coalesce(1)

python
df.coalesce(1).write.parquet(...)   # ⚠️

把所有分区并成 1 个 → 最后只剩 1 个 Task 在算，其他机器全闲着，变成单线程瓶颈。大数据很忌讳这个。

repartition = 打散重排（贵，有 Shuffle）；coalesce = 就近合并（省，尽量不 Shuffle）。

07综合例题：把这条链走一遍

python
a = orders.filter("order_status = 'delivered'")   # 窄依赖（原地过滤）
b = a.select("order_id", "customer_id")           # 窄依赖（选列）
c = b.repartition(8, "customer_id")               # ⚠️ 宽依赖 → Shuffle
d = c.groupBy("customer_id").count()              # ⚠️ 再次 Shuffle
e = d.coalesce(1)                                 # 合并成 1 个分区 → 单 Task 瓶颈

拆解：

  1. 窄依赖：filter、select
  2. 产生 Shuffle：repartition(8)、groupBy（两次）
  3. repartition(8, "customer_id") 做了什么：按 customer_id 哈希，把相同客户的数据集中到 8 个分区，为 groupBy 做准备
  4. coalesce(1) 的问题：只剩 1 个 Task，其他 Executor 空闲 → 性能瓶颈
  5. Stage 边界：在 repartition 和 groupBy 两次 Shuffle 处切 → 大约 3 个 Stage

一张图收尾：

text
orders(3个分区)
   │
   ├─ filter  ── 窄依赖，原地算
   ├─ select  ── 窄依赖，原地算
   │                 ┌─ Stage 0（不用等别人，并行跑）
   │
   ├─ repartition(8) ─ 宽依赖 → 搬家 → ● Shuffle ①
   │                 ┌─ Stage 1
   ├─ groupBy ─────── 宽依赖 → 搬家 → ● Shuffle ②
   │                 ┌─ Stage 2
   └─ coalesce(1) ── 就近合并（几乎不 Shuffle）→ 单 Task

08小结

这条线到这里就通了：Partition → 窄/宽依赖 → Shuffle → Stage → repartition/coalesce → Job。

核心一句话：窄依赖原地算、宽依赖要搬家（Shuffle）、Stage 以 Shuffle 为界翻篇、coalesce(1) 是单线程陷阱。

下一篇计划顺着“Stage 是系统划的界”继续往下钻：DAGScheduler 到底怎么把一张 DAG 切成 Stage、Shuffle 在源码里对应哪个类——把这篇的概念在 SparkPlan 和 ShuffleManager 的代码里对上号。