企业官网建设流程全解析

技术社区里聊Spark，RDD的血缘关系经常被一句"宽依赖有shuffle，窄依赖没有shuffle"带过。这句话没错，但真正做性能调优、排查任务失败、理解Stage划分的时候，会发现宽窄依赖背后还有不少值得掰开揉碎的东西。这篇就围绕Spark RDD依赖关系做一次深入拆解，重点讲清楚窄依赖和宽依赖的判定逻辑、它们如何影响Stage划分和故障恢复，以及日常调优中怎么利用依赖关系少走弯路，适合那些已经会写RDD算子但还没把调度机制串起来的同学。

1. 从血缘机制说起：RDD为什么需要依赖关系

1.1 依赖不是一个形式概念，它是RDD"可复盘"的地基

RDD不可变，但Spark需要知道一个RDD的分区数据是怎么来的。于是每个RDD内部会保存一组依赖（Dependencies），描述当前RDD的分区与父RDD分区之间的映射关系。这个映射关系就是血统（Lineage）。遇到某个executor崩溃、某个分区数据丢失时，Spark不需要闲着等副本，而是顺着依赖链重新计算丢失的那部分数据，这就是Spark自带容错的核心。

打个比方，这有点像乐高手册：你现在手里的RDD是拼好的某个城堡，依赖就是图纸里的分层视图。城堡坏了不用重新买整套乐高，按图纸把坏掉的那块捡起来对应步骤补上即可。如果没有依赖关系，RDD之间就是一堆孤立的零件，Spark既不知道任务往哪下发，也不知道数据丢了该从哪里重算。

1.2 依赖关系直接决定DAGScheduler的"切刀"

Spark在真正执行前有一层逻辑调度：DAGScheduler把作业拆成Stage。拆的关键依据不是"有几步算子"，而是依赖类型。碰到窄依赖就把转换算子尽量合并进同一个Stage，让数据在一个个分区上完成一系列函数；碰到宽依赖就在中间切一刀，形成Stage边界，因为数据需要跨节点交换。

也就是说，宽窄依赖不仅是数据血缘问题，还直接决定你的作业会产生多少个Stage、shuffle发生在哪里、task失败后恢复成本有多大。

1.3 宽窄依赖的定义：始终回到"父分区对应几个子分区"

判断宽窄的高频标准很简单：

• 窄依赖：父RDD的每个分区最多被子RDD的一个分区使用。
• 宽依赖：父RDD的每个分区可能被多个子分区使用，因此父分区的数据要按规则拆开、分发到下游不同节点。

只看map、filter这类算子很好判断。但真正容易混淆的是各种"看起来像一对多"的场景，比如coalesce、union、join。我这里先埋个引子，后面用专门一章讲具体算子的判定方法。

1.4 在源码层面对应两个大类

org.apache.spark.Dependency这个抽象类往下，Spark核心就两条路线：

• NarrowDependency：窄依赖。它下又有OneToOneDependency、RangeDependency等细分。
• ShuffleDependency：宽依赖。它携带了partitioner、aggregator、mapSideCombine这些和shuffle有关的配置。

后续无论看源码作者实现什么算子，你只要去问一句getDependencies返回的是哪种类型，就能判断这个算子的边界行为了。

2. 窄依赖：每个父分区至多一个子分区，Spark为什么偏爱它

2.1 窄依赖的几种具体形态

第一类最直觉，OneToOneDependency。map、flatMap、filter、mapPartitions这些算子都是父分区i的数据经过处理后进入子分区i，分区序号一一对应。任务执行时不需要看任何其他父分区，单分区内跑完一个函数，就能得到子分区。

第二类是RangeDependency。典型代表是union操作。两个父RDD连接起来，子RDD的前面一段分区来自父RDD1，后一段来自父RDD2，每个父分区只进入一个子分区。所以union在依赖关系上依然是窄依赖，不需要shuffle。

第三类容易被忽略，但很实用：多个父分区合并成一个子分区，但同时满足"每个父分区不会被分到多个子分区"。典型代表是coalesce(partitions, shuffle = false)。比如8个分区合并成4个分区，8个父分区分别归属到4个子分区，一个父分区不会同时进两个子分区，因此它依然属于窄依赖。

2.2 窄依赖如何支持pipeline式执行

窄依赖下，从读入到输出，数据可以在同一个节点的同一个Task里连续加工完。Stage内部的所有窄依赖算子会被DAGScheduler打包成一组任务并发执行。每个task执行时会先执行这个stage的最后一个rdd，再按依赖链回溯执行父RDD算子？准确说，是沿着分区一级级向下计算。

你可以把窄依赖想象成流水线：一个零件进流水线，经过一系列工位，每个工位只处理自己手里那个零件，不需要和其他产线交换半成品。因为每个分区计算彼此独立，Spark可以最大化并行度，一次task里完成多个窄依赖转换，性能自然好。

在Spark UI的Stage执行图里，一段窄依赖链条通常呈现为一个小方框内的多个操作块，这个方框就是一个Stage。你很少看到窄依赖导致慢任务，因为它没有跨节点IO。

2.3 窄依赖在容错上的便宜之处

如果窄依赖链上的某个分区计算失败，Spark不需要重跑整个RDD的全部分区，只重跑丢失分区对应的父分区即可。恢复范围精确到分区粒度。

举个真实场景。某个作业上游读HDFS，中间经过map、filter、mapPartitions三类窄依赖，某台worker宕机导致最后3个分区task失败。由于依赖是one-to-one的，Spark只需要把这三张父分区映射对应的输入重新算一遍，其他分区完全不受影响，很快可以恢复。

如果是宽依赖呢？情况就不同了。一个父分区的数据可能已经被拆成多份发到下游N个分区，一旦有一个父分区对应的shuffle输出丢失或损坏，很可能需要重新计算上游Stage并重新shuffle。恢复成本呈数量级上升。

2.4 别看到"分区变少"就误判成宽依赖

现实中有人把coalesce当成"一定会shuffle"的操作，其实默认情况下恰恰相反。coalesce(200, shuffle = false)是把当前Executor内已有的分区合并，不发生跨节点传输；coalesce(200, shuffle = true)才会上演repartition，也就是ShuffleDependency。

判断口径始终只有一个：一个父分区的数据会不会被切分成多份，同时送往多个不同子分区。只要不会，哪怕子分区数量变少，也是窄依赖。

3. 宽依赖：一旦出现ShuffleDependency，Stage就在你眼前裂开

3.1 宽依赖的本质是数据重分布

宽依赖的场景往往是需要"按key重排"。比如groupByKey，需要把相同key的所有value聚集到同一个下游分区。上游某个分区中散落着多个key的数据，它们会流向不同的下游分区。这就天然是一对多关系。

因为数据跨节点流动，宽依赖在物理执行上必然转化为shuffle。每个上游task会按分区器（通常默认reduceByKey用HashPartitioner）计算每条记录的目标分区，写入本地磁盘上的shuffle文件；下游task再去拉取属于自己分区的数据文件，这就是Shuffle Read/Write。

一句话：宽依赖是逻辑上的血缘标记，shuffle是它在物理世界的落地方式。

3.2 Stage划分原理：从后往前切，一遇shuffle就停

DAGScheduler在提交作业时，会从final RDD开始反向遍历依赖树。当遇到一个ShuffleDependency，它不会继续往前合并成一个Stage，而是把这个shuffle作为边界：shuffle之前的父RDD们归入一个"上游Stage"，shuffle之后的RDD们归入"下游Stage"。

这就是为什么你在UI上看到一个作业被拆成多个Stage时，Stage之间总会有一个"shuffle"的标记线。窄依赖可以一直往回串，直到串到数据源或某个shuffle为止。

从源码看，getShuffleDependencies和getParentStages就是在做这件事：找齐所有不经过shuffle的祖先RDD，作为当前Stage的一部分；再向上找所有需要shuffle的父Stage，逐个提交。

3.3 宽依赖可能带来的数据倾斜和额外开销

宽依赖的开销来自几个方面：shuffle write要写磁盘、shuffle read要拉网络数据、序列化反序列化、以及可能的内存溢出排序。如果数据分布不均匀，最常见的就是某个下游分区拿到超大key集合，导致几个task特别慢，这就是数据倾斜。

实际操作中，groupByKey和reduceByKey都是ShuffleDependency，但推荐用后者，因为reduceByKey会在map端做本地合并，先聚合一部分value再shuffle，传输数据量小得多。不过这不改变它仍是宽依赖的事实，只是宽依赖里的"搬运量"变少了。

3.4 宽依赖的血缘惩罚：重算不是"分区级"而是"stage级"

窄依赖丢了一个分区，重算这个分区对应父分区就行。宽依赖丢失某个分区后，即使只是下游一个分区算错了，也可能让上游整个Stage重跑一遍，因为shuffle输出文件是整体依赖关系的一部分。

尤其是多级宽依赖的长作业，比如A -> shuffle -> B -> shuffle -> C，一旦中间环节出了情况，重算链路会拉得非常长。这也是为什么checkpoint会被反复推荐：在宽依赖之后落一次盘，切断太长的血缘链条，下次恢复直接从checkpoint点继续，而不用从源头重算。

我自己处理过类似问题：一个跑了30分钟的任务在最后的shuffle读阶段偶发失败，重试一次往往要再等20分钟。后来在中间结果RDD上加了checkpoint，重试时间压到了3分钟以内。这就是宽依赖容错代价造成的明显差异。

4. 一个算子一个坑：常见算子的宽窄依赖判定表

4.1 先给一张可以直接抄的判定表

4.2 reduceByKey和groupByKey：都是宽依赖，但性能差很多

很多人误以为reduceByKey是窄依赖，因为它比groupByKey快。实际上它们都触发shuffle。差别在于reduceByKey在shuffle前先在每个map端task里做一次局部合并：同一个task内存里相同key先加一遍，再写shuffle文件。groupByKey则是把原样数据全部打散，shuffle read后再做group。

所以在你不需要"每个key对应的完整value集合"时，务必选reduceByKey或aggregateByKey。不要贪图groupByKey的直观。

4.3 join的宽窄是"可变"的：分区器决定一切

join是面试和实战里最容易让人晕的点。普通两个未分区RDD做join，必然会产生ShuffleDependency，需要把相同key从不同节点拉在一起。但如果两个RDD都已经用同一个Partitioner、同一个分区数预先分好区，那么子RDD分区和父RDD分区保持一一对应，join就可以变成窄依赖。

实战中可以通过partitionBy(new HashPartitioner(200))对两个RDD做一次预处理，之后多次join都不再shuffle，代价只是首次partitionBy会shuffle一次。这很适合"维度表+大表反复join"的场景。

我自己做过一个实验：两张大约千万级key的表join，不分区的普通join会有一个明显的Shuffle Read阶段；两个都先partitionBy再去join，UI上shuffle阶段的整体耗时消失了，作业总时间缩短了接近40%。当然这是数据分散度较好、分区数合理的情况。

5. 在Spark UI和日志里定位宽依赖：肉眼侦察shuffle位置

5.1 DAG图：两个Stage之间的那道缝就是宽依赖

Spark UI的Job页会画出一个DAG图：每个方框就是一个Stage，方框里面的小圆圈代表窄依赖算子；方框之间用连接线串起来，这条线就是shuffle/宽依赖。

看UI时我一般先数Stage数量。Stage数量越多，说明shuffle次数越多。再点进某个Stage，看它的"Shuffle Read Size / Records"和"Shuffle Write Size"，就能定位到数据流量最大的宽依赖位置。

如果发现某个Stage的Shuffle Read特别大，而后续Stage又没做多少有效计算，基本说明上游宽依赖打散了太多数据，考虑map端预聚合或分区策略优化。

5.2 用toDebugString快速打印血缘

命令行加载完RDD后，调一下rdd.toDebugString，Spark会打印出一棵依赖树。看到ShuffledRDD出现的位置，就相当于看到了宽依赖。

以wordcount为例：

val rdd = sc.textFile("hdfs:///logs/") .flatMap(_.split(" ")) .map(w => (w, 1)) .reduceByKey(_ + _) rdd.toDebugString

输出大致是这样：

(2) MapPartitionsRDD[4] at reduceByKey at test.scala:8 [] | ShuffledRDD[3] at reduceByKey at test.scala:8 [] +-(2) MapPartitionsRDD[2] at map at test.scala:7 [] | MapPartitionsRDD[1] at flatMap at test.scala:6 [] | ParallelCollectionRDD[0] at textFile at test.scala:5 []

ShuffledRDD就是宽依赖的实体。它上面的MapPartitionsRDD[4]是reduceByKey的后续结果RDD，窄依赖连接到它；下面的map、flatMap都属于同一个上游Stage，窄依赖连接。

5.3 日志与task重试信息里的宽依赖痕迹

当Spark因为executor丢失重跑task时，日志里往往会出现"resubmitting task"或"lost shuffle outputs"相关关键字。如果一行报错后面跟着"FetchFailed"，通常就是shuffle读阶段出问题，也就是某个宽依赖环节整体重算。

排查思路：先看UI中失败Stage处于哪个依赖边，再决定是调整分区、加内存、还是加checkpoint。不要盲目提升spark.task.maxFailures，那样只会让宽依赖重算成本再炸一次。

6. 我实践中的宽依赖优化经验与常见误区

6.1 优先减少shuffle数据量，而不是消灭shuffle本身

宽依赖在绝大多数场景下是无法彻底避免的。与其追求"零shuffle"这种漂亮口号，不如先把shuffle搬运的数据量压下去。

最直接的方法是map端合并：能用reduceByKey就不用groupByKey，能用aggregateByKey带上初始值就不用手动map再group。另一个有效招数是"加盐+两阶段聚合"处理热点key。先给key加随机前缀，做一次局部聚合，去掉前缀后再做第二次聚合。这个操作会引入额外的一次shuffle，但能把单点task的压力打散，整体执行时间和稳定性提升反而更明显。

6.2 谨慎使用repartition和coalesce

repartition就是coalesce(shuffle=true)，它会无条件产生一次全量shuffle。如果目的是增加分区数量，那是没有办法的办法；如果目的是减少分区数量，优先用coalesce(shuffle=false)，因为它能复用已有节点上的分区，不产生shuffle。

很多人不小心在算子末尾加repartition(n)只是为了控制后续输出文件数量，结果白白多一次shuffle。正确做法是尽量结合coalesce或调整上游写入时的分区方式。

6.3 checkpoint是宽依赖长血统的救命稻草

我之前处理过一个嵌套多层宽依赖的离线任务，血缘链长到"跑两小时，一遇到节点故障就要从入口重算"。后来在中间结果处加checkpoint：

val midRdd = source.map(...).reduceByKey(...) midRdd.checkpoint() midRdd.count() // 触发checkpoint写入

这么做的意义是切断过长的血缘，让后续计算从checkpoint保存的RDD重新开始，而不是追溯到底层数据源。注意checkpoint需要调用一次action才会真正落盘，这也是很多人踩过的坑。

6.4 不要被"这个算子写出来没有shuffle"骗了

有些算子在UI里看起来没shuffle，但底层仍然会发生数据重分区，只是被优化器藏起来了。比如某些版本里groupBy的默认行为可能先做一次partitionBy；再比如sortBy看起来只是orderBy，但底层默认走ShuffledRDD。

所以在判断依赖类型时，我更倾向于直接相信toDebugString和UI上的Stage边界，而不是凭直觉。action一跑，有没有Shuffle Read/Write一目了然，这比背算子表可靠得多。

6.5 关于广播变量替代宽依赖的经验

如果两个RDD做join，其中一个小RDD的量级只有几十MB，完全可以用广播方式处理。先把小表collect成Map，广播出去，然后在大表上做map端匹配。这样原本join里最重的ShuffleDependency就被替换成了窄依赖加广播读，执行时间往往能断崖式下降。

但广播也会有一个代价：driver内存和每个executor内存都要承受一份广播副本。所以一旦小表变大，这条路的性价比就会降低，反而普通的shuffle join更稳定。诀窍是在代码里写清楚阈值，并判断小表是否真的能装进广播。

做Spark调优做得久了你会发现，宽窄依赖不是单纯的数据结构概念，它和你线上的每次慢任务、每次失败恢复、每次Stage分割都绑定在一起。真正值钱的不是记住"哪个算子宽、哪个算子窄"，而是遇到一个陌生算子时，能够通过依赖树的特征、Stage边界的位置、shuffle指标的变化，反推回真正的数据流转路径。掌握了这套方法，再看Spark UI和分析日志，整个作业就像被拆成了透明积木一样，哪里慢、哪里要重试、哪里该加checkpoint，心里基本有数。