宽依赖与窄依赖：Stage 是怎样划分出来的

上一篇拆解了 RDD 的五大属性，其中 dependencies 决定了 RDD 之间的依赖类型。这一篇进入依赖类型的核心区分——宽依赖和窄依赖——以及它如何直接决定 Stage 的划分。

宽依赖和窄依赖容易被理解成"一对一"和"多对多"的分区关系。更准确的说法是：窄依赖意味着子 RDD 的每个分区只依赖父 RDD 的固定少数分区，可以在单个 Task 内完成计算；宽依赖意味着子 RDD 的分区需要读取父 RDD 所有分区的数据，必须等待一次全量数据交换（Shuffle）。

本文只抓一个问题：DAGScheduler 按什么规则把 RDD 依赖图切分成 Stage。

窄依赖的三种形态

窄依赖（NarrowDependency）的定义是：父 RDD 的每个分区最多被子 RDD 的一个分区使用。在 Spark 源码中，NarrowDependency 有两个具体子类：

OneToOneDependency：父子分区一一对应。map、filter、flatMap 产生这种依赖。父 RDD 的第 i 个分区只被子 RDD 的第 i 个分区使用。

RangeDependency：父 RDD 的一段连续分区对应子 RDD 的一段连续分区。union 操作产生这种依赖——把多个 RDD 的分区拼接在一起，每个父分区仍然只被一个子分区使用。

还有一种特殊情况：如果两个 RDD 有相同的 Partitioner（co-partitioned），它们的 join 也是窄依赖。因为相同 key 已经在相同的分区号中，不需要数据重分布。

窄依赖示例：

  父 RDD                子 RDD
  ┌────┐               ┌────┐
  │ P0 │───────────────>│ P0 │   OneToOne (map/filter)
  ├────┤               ├────┤
  │ P1 │───────────────>│ P1 │
  ├────┤               ├────┤
  │ P2 │───────────────>│ P2 │
  └────┘               └────┘

  父 RDD-A   父 RDD-B        子 RDD
  ┌────┐     ┌────┐         ┌────┐
  │ P0 │────────────────────>│ P0 │   Range (union)
  ├────┤     │    │         ├────┤
  │ P1 │────────────────────>│ P1 │
  └────┘     ├────┤         ├────┤
             │ P0 │────────>│ P2 │
             └────┘         └────┘

窄依赖的核心特征是：计算子分区时只需要读取父 RDD 的一个（或固定少数个）分区的数据，不需要跨网络拉取其他节点上的数据。

宽依赖：Shuffle 的入口

宽依赖（ShuffleDependency）的定义是：父 RDD 的每个分区可能被子 RDD 的多个分区使用。这意味着父 RDD 的数据需要按 key 重新分配到不同的分区，也就是 Shuffle。

典型的宽依赖算子：

groupByKey    → 按 key 分组，每个 key 的数据可能来自所有父分区
reduceByKey   → 先 map 端局部聚合，再 Shuffle 到 reduce 端全局聚合
join          → 如果两个 RDD 的 Partitioner 不同，需要重分布数据
repartition   → 显式重新分区，必然触发 Shuffle
sortByKey     → 全局排序需要 RangePartitioner，触发 Shuffle

宽依赖的数据流是全连接的：

父 RDD                子 RDD
┌────┐     ┌─────>   ┌────┐
│ P0 │─────┤         │ P0 │
├────┤     └─────>   ├────┤
│ P1 │─────┬─────>   │ P1 │
├────┤     └─────>   ├────┤
│ P2 │───────────>   │ P2 │
└────┘               └────┘

每个父分区的数据按 key 的 hash 值分散到所有子分区

宽依赖意味着子 RDD 的一个分区需要读取父 RDD 所有分区中属于自己的那部分数据。这个"读取"操作需要网络传输（Shuffle Read），而且必须等父 RDD 的所有分区都计算完毕并写出 Shuffle 文件（Shuffle Write）后才能开始。

Stage 划分算法

DAGScheduler 的 Stage 划分算法可以用一句话概括：从 action 触发的最终 RDD 开始，沿着依赖链向前回溯；遇到窄依赖就合并到当前 Stage，遇到宽依赖就切出一个新 Stage。

具体流程：

1. 从最终 RDD 开始，创建一个 ResultStage
2. 遍历该 RDD 的 dependencies：
   - 如果是 NarrowDependency，把父 RDD 加入当前 Stage
   - 如果是 ShuffleDependency，为父 RDD 创建一个新的 ShuffleMapStage
3. 对每个新创建的 ShuffleMapStage 递归执行步骤 2
4. 直到所有 RDD 都被分配到某个 Stage

这个算法的结果是：一个 Job 被切成一棵 Stage 树。叶子节点是读取数据源的 Stage，根节点是产出最终结果的 ResultStage。Stage 之间的边就是 Shuffle。

举一个完整的例子：

val lines = sc.textFile("data.txt")      // RDD-0
val words = lines.flatMap(_.split(" "))   // RDD-1  (窄依赖)
val pairs = words.map(w => (w, 1))        // RDD-2  (窄依赖)
val counts = pairs.reduceByKey(_ + _)     // RDD-3  (宽依赖 ← Shuffle 边界)
val filtered = counts.filter(_._2 > 10)  // RDD-4  (窄依赖)
filtered.collect()                        // Action，触发 Job

Stage 划分结果：

Stage 0 (ShuffleMapStage):
  RDD-0 (textFile) → RDD-1 (flatMap) → RDD-2 (map)
  输出: Shuffle 文件

        ← Shuffle 边界 →

Stage 1 (ResultStage):
  RDD-3 (reduceByKey) → RDD-4 (filter)
  输出: collect 结果返回 Driver

Stage 0 包含了 textFile → flatMap → map 三个窄依赖变换，可以在同一个 Task 内 pipeline 执行。Stage 1 包含了 reduceByKey → filter 两个操作。Stage 1 必须等 Stage 0 所有 Task 完成并写出 Shuffle 文件后才能启动。

Stage 内部的 Pipeline 执行

窄依赖被合并到同一个 Stage 的好处是可以 pipeline 执行。Stage 0 中的 textFile → flatMap → map 三个变换不会产生三轮独立的迭代。相反，每个 Task 把三个变换串成一个迭代器链：

Task 处理一条记录的路径：

  从 HDFS 读一行 (textFile)
      │
      ▼
  拆成单词数组 (flatMap)
      │
      ▼ (对每个单词)
  变成 (word, 1) (map)
      │
      ▼
  写入 Shuffle 缓冲区

整个过程中，一条记录从读取到写入 Shuffle 缓冲区，中间数据不需要物化到磁盘或全量加载到内存。这就是 DAG 模型相对 MapReduce 减少 I/O 的关键机制。

实验：观察 Stage 划分

在 spark-shell 中构造不同依赖类型的计算链，观察 Stage 数量：

// 纯窄依赖链：只有 1 个 Stage
val rdd1 = sc.textFile("README.md")
  .flatMap(_.split(" "))
  .map(_.toLowerCase)
  .filter(_.length > 3)
rdd1.count()
// Spark UI: 1 个 Stage

// 一次 Shuffle：2 个 Stage
val rdd2 = sc.textFile("README.md")
  .flatMap(_.split(" "))
  .map(w => (w, 1))
  .reduceByKey(_ + _)
rdd2.count()
// Spark UI: 2 个 Stage

// 两次 Shuffle：3 个 Stage
val rdd3 = sc.textFile("README.md")
  .flatMap(_.split(" "))
  .map(w => (w, 1))
  .reduceByKey(_ + _)     // Shuffle 1
  .map { case (w, c) => (c, w) }
  .sortByKey()             // Shuffle 2
rdd3.collect()
// Spark UI: 3 个 Stage

打开 Spark UI（默认 http://localhost:4040），在 Jobs 页面点击某个 Job，可以看到 Stage 的 DAG 可视化。每个 Stage 显示为一个蓝色方框，Stage 之间的 Shuffle 显示为黑色箭头。

宽依赖的代价

宽依赖引入的 Shuffle 有三个层面的代价：

计算代价：Shuffle Write 阶段，每个 Map Task 需要把输出数据按 key 的 hash 值分成 R 份（R 是下游分区数），排序后写入磁盘。Shuffle Read 阶段，每个 Reduce Task 需要从所有 Map Task 的输出中拉取属于自己的那份数据。

I/O 代价：Shuffle 数据必须写入本地磁盘（Shuffle Write 的输出是磁盘文件，不是内存缓冲区），然后通过网络传输到下游 Executor，再从磁盘读取。整个过程至少涉及一次磁盘写入、一次网络传输、一次磁盘读取。

同步代价：Shuffle 是一个全局同步点。下游 Stage 的任何 Task 都不能在上游 Stage 的所有 Task 完成之前启动。如果上游某个 Task 特别慢（数据倾斜或硬件故障），整个下游 Stage 都被阻塞。

这三个代价是 Spark 性能调优中最常需要关注的。后续第 04 篇会详细展开 Shuffle 的内部机制和优化手段。

模式提炼

Stage 划分体现了一个通用模式：在同步边界处切分执行计划。

任何分布式计算系统都需要在某些点上进行全局数据交换（Shuffle/Exchange/Repartition）。这些点天然构成了执行计划的分界线：分界线内的操作可以本地执行、pipeline 串联；分界线处必须等待全局同步。

数据库查询引擎中的 Exchange 算子扮演了相同的角色：它把查询计划切成可以在单个节点上独立执行的片段（Fragment），Fragment 之间通过 Exchange 交换数据。Flink 中的 Operator Chain 按同样的逻辑工作：可以 chain 在一起的算子合并到一个 Task 中执行，需要数据重分布的地方断开 chain。

工程迁移表

Spark 概念	MapReduce 对应	Flink 对应	数据库对应
窄依赖	无显式概念（Map 内部的组合）	Forward/Rescale 连接模式	同一 Fragment 内的算子管道
宽依赖（ShuffleDependency）	Map 到 Reduce 之间的 Shuffle	Keyby() 后的 Hash 分区	Exchange 算子
Stage	Map 阶段 / Reduce 阶段（固定两个）	Operator Chain（按 Shuffle 切分）	查询计划的 Fragment
Pipeline 执行	无（Map 输出全部落盘后 Reduce 才开始）	同一 Chain 内的算子融合	同一 Fragment 内的 Volcano 迭代
ShuffleMapStage	Map 阶段	需要做 Shuffle 的上游 SubTask	产出 Exchange 数据的子计划
ResultStage	Reduce 阶段	不需要下游 Shuffle 的末端 SubTask	返回结果的顶层子计划

常见误解

误解一：“Stage 数量等于 Shuffle 次数”。Stage 数量 = Shuffle 次数 + 1。一次 Shuffle 产生两个 Stage（上游 ShuffleMapStage 和下游 Stage），但多次 Shuffle 的中间 Stage 被复用。准确地说，N 次 Shuffle 产生 N+1 个 Stage。

误解二：“窄依赖一定不需要网络传输”。窄依赖不需要 Shuffle，但如果父 RDD 的分区缓存在另一台机器上，读取缓存数据仍然需要网络传输。窄依赖避免的是全量数据重分布（Shuffle），不是所有网络通信。

误解三：“repartition 和 coalesce 都是宽依赖”。repartition 是宽依赖（触发 Shuffle），但 coalesce 在减少分区数时使用窄依赖——它把多个父分区合并到一个子分区中，不需要数据重分布。只有增加分区数时 coalesce(n, shuffle=true) 才触发 Shuffle。

练习

构造一个包含 join 的计算：两个 RDD 分别 reduceByKey 后 join。用 toDebugString 观察总共有几个 Stage。然后让两个 RDD 在 reduceByKey 时使用相同的 Partitioner（相同分区数），观察 Stage 数量是否变化。
对一个 RDD 分别调用 repartition(10) 和 coalesce(10)（假设原始分区数大于 10），用 dependencies 检查产生的依赖类型。再对原始分区数小于 10 的 RDD 调用 coalesce(10)，观察行为变化。
在 Spark UI 的 Stage 详情页面中，找到 “Shuffle Read” 和 “Shuffle Write” 指标。对比不同 Shuffle 算子（groupByKey vs reduceByKey）的 Shuffle Write 数据量差异，思考 map 端预聚合对 Shuffle 数据量的影响。

系列导航

序号	主题	状态
00	导读：为什么 Spark 的核心是一张 DAG	已完成
01	RDD：弹性分布式数据集的五大属性	已完成
02	宽依赖与窄依赖：Stage 是怎样划分出来的	本篇
03	DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行	下一篇
04	Shuffle 机制：数据跨分区交换的代价与优化
05	内存管理与 Tungsten：堆外内存、序列化与代码生成
06	存储体系：Block Manager、广播变量与累加器
07	容错机制：Lineage、Checkpoint 与推测执行
08	Spark SQL 与 Catalyst：从 SQL 文本到物理算子树
09	DataFrame 与 Dataset：类型安全与性能的折中
10	Adaptive Query Execution：运行时改写执行计划
11	数据源 API 与谓词下推：让存储层帮忙过滤
12	Structured Streaming：把流当成无界表
13	水位线与状态管理：流处理的时间语义
14	Spark on YARN / K8s：资源管理与部署模式
15	动态资源分配与作业调度：多租户集群的资源博弈
16	性能诊断：Spark UI、Event Log 与调优思路
17	Spark 与 Flink：两种计算引擎的设计选择

参考资料

Matei Zaharia et al. Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing. NSDI 2012. Section 4: Stage 划分与调度。
Apache Spark 源码：core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala https://github.com/apache/spark
Apache Spark 源码：core/src/main/scala/org/apache/spark/Dependency.scala
Apache Spark 官方文档：Cluster Mode Overview. https://spark.apache.org/docs/latest/cluster-overview.html