1. 从一个翻车现场说起为什么要搞懂宽窄依赖刚接触 Spark 那会儿我写过一段看起来特别正常的代码从 HDFS 读一份日志做一次groupByKey再map一下格式化输出。逻辑上没毛病跑起来也没报错但那天集群上同时跑了七八个任务我这段代码一上去整个 Stage 卡了将近四十分钟其他任务全在排队。后来把 DAG 图拉出来一看问题出在一次不必要的 shuffle 上——我本可以用reduceByKey在 map 端先做一次局部聚合却偏偏用了groupByKey把全量数据拉到一起。这就是典型的“不懂宽窄依赖写得出能跑的代码写不出跑得快的代码”。Spark 里 RDD 的依赖关系分两种窄依赖Narrow Dependency和宽依赖Wide Dependency。这不是一个学术概念它直接决定了你的任务会被切成几个 Stage、会不会触发 shuffle、数据要不要跨节点网络传输、失败重算的代价有多大。说白了宽窄依赖就是 Spark 调度层的“分水岭”理解了它你才能看懂 DAG 图、才能判断一段代码到底贵在哪、才能知道为什么有些算子天生就慢。这篇文章适合三类人一是刚学 Spark、能写map/filter但说不清 Stage 怎么划分的二是写过一段时间、遇到过数据倾斜和 shuffle 慢但不知道从哪下手的三是准备面试、被问到“宽窄依赖区别”只能背概念却讲不出所以然的。我会从依赖的本质讲起把算子分类、Stage 切分、shuffle 代价、失败恢复、实操验证这几块串起来尽量让你读完能自己动手把 DAG 图画出来、把 Stage 数出来。2. 宽窄依赖的本质父 RDD 的每个分区到底被谁用了2.1 一句话定义与判断标准判断宽窄依赖只需要问一个问题父 RDD 的每一个分区是不是只被一个子 RDD 分区使用如果父 RDD 的每个分区最多只被子 RDD 的一个分区依赖那就是窄依赖。如果父 RDD 的某个分区被子 RDD 的多个分区依赖那就是宽依赖。这个定义听起来抽象但换成生活场景就好懂了。窄依赖像“一对一配送”一个仓库父分区的货只发给一个门店子分区路线固定谁缺货补谁就行。宽依赖像“重新洗牌分拣”所有仓库的货先集中到一个分拣中心按新规则重新打包再发给各个门店——这个“集中再分发”的过程就是 shuffle。窄依赖的典型算子有map、filter、flatMap、mapPartitions、union、coalesce不开启 shuffle 时。宽依赖的典型算子有groupByKey、reduceByKey、sortByKey、join非同一分区器时、repartition、distinct、intersection。2.2 为什么这个判断标准如此重要因为 Spark 的调度器就是靠这个标准来切 Stage 的。遇到宽依赖就切一刀形成一个 Stage 边界。一个 Stage 内部全是窄依赖可以流水线pipeline执行数据在内存里一路算下去不需要落盘、不需要跨节点而宽依赖必须等上游所有分区的数据都准备好才能开始 shuffle。我常跟人说窄依赖是“可以偷懒的依赖”宽依赖是“必须等的依赖”。窄依赖下如果某个分区数据丢了Spark 只需要从对应的那一个父分区重算代价可控宽依赖下如果下游某个分区丢了可能要把上游所有分区都重算一遍代价高得多。这就是为什么 Spark 把宽依赖的中间结果尽量物化到磁盘——不是它想写磁盘是它不敢不写。2.3 一张表看清两类依赖的差异对比维度窄依赖宽依赖分区对应关系父分区到子分区一对一或多对一父分区到子分区一对多是否触发 shuffle否是Stage 划分Stage 内部Stage 边界失败重算代价只重算对应父分区可能重算全部上游分区数据是否跨节点通常不跨除非分区本就分布必然跨节点网络传输典型算子map、filter、uniongroupByKey、join、repartition执行方式流水线并行需等待上游全部完成这张表建议你记牢面试和排查问题时都用得上。尤其是“失败重算代价”这一行很多人只记得宽依赖会 shuffle却忽略了它在容错上的代价这在实际生产里影响很大。3. 算子分类实操哪些算子会偷偷触发宽依赖3.1 一眼看穿的窄依赖算子map、filter、flatMap这三个是最纯粹的窄依赖父分区怎么分子分区就怎么分一个不多一个不少。mapPartitions和mapPartitionsWithIndex也是窄依赖只不过操作粒度从单条记录变成了整个分区适合做分区级别的初始化比如建数据库连接。union比较特殊它把两个 RDD 合在一起父 RDD 的分区直接变成子 RDD 的分区没有数据移动所以是窄依赖。coalesce在不开启 shuffle 参数时也是窄依赖它只是把相邻分区合并比如 4 个分区合成 2 个本质是减少分区数而不打乱数据。这里有个容易踩的坑coalesce默认不 shuffle但如果你把分区数调大它其实不会生效因为窄依赖没法把一个分区拆成多个。想增加分区必须用repartition而repartition是宽依赖。我见过有人写coalesce(100)想把分区从 10 扩到 100结果分区数纹丝不动还纳闷为什么并行度上不去。3.2 披着窄依赖外衣的“伪窄依赖”mapValues和flatMapValues只作用于键值对的 value不改变 key所以不触发 shuffle是窄依赖。但很多人一看名字里有“Values”就以为要聚合其实它跟map一样轻量。sample和filter类似按比例抽样不改变分区结构窄依赖。glom把每个分区的数据聚成一个数组也是窄依赖。真正需要警惕的是sortByKey、groupByKey、reduceByKey、join、cogroup、distinct、intersection、subtract、repartition、repartitionAndSortWithinPartitions这些。它们要么需要按 key 重新分布数据要么需要全局排序要么需要去重本质上都要求“相同 key 的数据聚到一起”而相同 key 的数据原本可能散落在任意分区所以必须 shuffle。3.3 一个反直觉的例子join 不一定都是宽依赖这是很多人忽略的点。如果两个 RDD 在 join 之前已经用相同的分区器分好了区并且分区数一致那么join可以变成窄依赖Spark 会做 map-side join不触发 shuffle。这就是所谓的“分区器一致优化”。举个例子RDD A 用partitionBy(new HashPartitioner(10))分过区RDD B 也用同样的分区器分过区那么A.join(B)时相同 key 一定在相同编号的分区里Spark 直接对应分区做 join 即可不需要再 shuffle。这个优化在宽表关联场景里非常值钱能省掉一次全量 shuffle。反过来如果两个 RDD 分区器不同或者其中一个没分区器那join就是宽依赖必须 shuffle。所以我在写多表关联时会刻意让参与 join 的 RDD 用同一个分区器尤其是大表 join 大表的场景。4. Stage 切分与 DAG宽依赖是怎么把任务切成段的4.1 从代码到 DAG 的映射过程你写的每一行 RDD 转换Spark 都会记在血缘Lineage里形成一张有向无环图也就是 DAG。DAG 不是给你看的装饰品它是调度器真正执行时依据的蓝图。切分规则很简单从后往前推遇到窄依赖就继续往前串遇到宽依赖就断开形成一个 Stage。断开的那个点就是 shuffle 的边界。上游 Stage 负责把数据按目标分区规则写出去下游 Stage 负责把数据读进来继续算。我习惯把 Stage 想象成“接力赛的交接区”。窄依赖是同一棒选手自己跑不用交接宽依赖是必须把棒子交给下一棒交接的时候所有人都得停下来等。交接区越多整体耗时越长因为每次交接都有等待和传输成本。4.2 一个具体例子的 Stage 划分假设有这样一段逻辑rdd sc.textFile(hdfs://path/logs) # Stage 0 起点 words rdd.flatMap(lambda line: line.split()) # 窄依赖 pairs words.map(lambda w: (w, 1)) # 窄依赖 counts pairs.reduceByKey(lambda a, b: a b) # 宽依赖切一刀 result counts.filter(lambda x: x[1] 100) # 窄依赖 result.saveAsTextFile(hdfs://path/out)这里会切成两个 Stage第一个 Stage 包含textFile、flatMap、map以及reduceByKey的 map 端局部聚合部分第二个 Stage 包含reduceByKey的 reduce 端合并、filter和saveAsTextFile。中间那次reduceByKey就是 shuffle 边界。如果你把reduceByKey换成groupByKeyStage 数量一样但第一个 Stage 不会做局部聚合所有数据原封不动地 shuffle 到下游网络传输量和内存压力会大很多。这就是为什么我一直强调能用reduceByKey就别用groupByKey前者在 map 端先做了一次合并相当于把宽依赖的代价提前削掉一部分。4.3 Stage 数量对性能的直接影响Stage 越多意味着 shuffle 次数越多磁盘 I/O 和网络传输越多任务整体延迟越高。但也不是说 Stage 越少越好——如果为了减少 Stage 而强行把宽依赖改成窄依赖可能导致逻辑错误或数据不完整。我的经验是先保证逻辑正确再优化 Stage 数量。优化手段包括合并多次 shuffle 为一次比如用repartitionAndSortWithinPartitions代替先repartition再sortByKey、用reduceByKey替代groupByKey、用mapSideCombine减少传输量、让 join 的 RDD 共享分区器。5. Shuffle 的代价宽依赖到底贵在哪里5.1 磁盘 I/O、网络传输与序列化三重开销宽依赖触发 shuffleshuffle 的过程分两半map 端写和 reduce 端读。map 端把数据按目标分区规则写到本地磁盘可能先写内存缓冲满了溢写reduce 端再通过网络把属于自己的那部分拉过来。这一写一读就带来了三重开销。第一是磁盘 I/O。map 端写临时文件reduce 端读临时文件数据量大的时候磁盘成为瓶颈。第二是网络传输。reduce 端要从多个 map 端拉数据跨节点传输带宽打满就卡住。第三是序列化与反序列化。数据在传输前要序列化成字节流到达后要反序列化成对象序列化框架选得不好CPU 开销会非常明显。我做过一个粗略的对比测试同样一份 100GB 的日志纯窄依赖的mapfilter链路跑完大概 8 分钟中间加一次groupByKey时间直接涨到 35 分钟以上。多出来的时间几乎全花在 shuffle 上。这不是 Spark 慢是宽依赖本身的物理代价摆在那里。5.2 数据倾斜宽依赖最凶险的副作用宽依赖还有一个更隐蔽的问题数据倾斜。因为宽依赖要按 key 重新分布数据如果某些 key 的数据量特别大对应的 reduce 分区就会成为“长尾任务”其他分区早就跑完了它还在慢慢磨。典型场景是groupByKey或join时某个热点 key 占了总数据量的 30% 甚至更多。这时候整个 Stage 的完成时间由最慢的那个分区决定并行度再高也没用。应对数据倾斜的常见手段有几种。一是加盐打散给热点 key 拼接随机前缀先做一次局部聚合再去掉前缀做全局聚合。二是广播小表如果是 join 导致的倾斜且其中一张表足够小可以用广播变量把整张小表发到每个 executor做 map-side join彻底避开 shuffle。三是提高并行度增加 shuffle 分区数让每个分区数据量小一点但这对极端热点 key 效果有限。提示数据倾斜不是宽依赖的错是数据分布本身的错。宽依赖只是把这个问题暴露出来了。窄依赖链路里如果有热点同样会慢只是不会像 shuffle 那样集中爆发。5.3 失败重算宽依赖的容错代价窄依赖下某个分区计算失败Spark 只需要从对应的父分区重新算一遍影响范围小。宽依赖下下游 Stage 的某个分区失败可能需要重新拉取上游所有 map 端输出如果上游数据已经不在内存或磁盘上甚至要重算整个上游 Stage。这就是为什么 Spark 在宽依赖边界会把中间结果尽量持久化。你可以手动调用persist或cache把某个 RDD 缓存起来避免重复计算。但缓存本身也占内存缓存策略要结合数据大小和内存容量来定。我一般会在两种情况下主动缓存一是某个 RDD 被多个下游分支复用二是某个 RDD 计算代价很高且后续会多次访问。缓存级别从MEMORY_ONLY到MEMORY_AND_DISK按需选内存不够就落盘别硬扛。6. 动手验证用 Spark UI 和代码把依赖关系看清楚6.1 用 toDebugString 打印血缘最直接的验证方式是用toDebugString。它会把 RDD 的血缘关系打印出来宽依赖的地方会有明显的标记。rdd sc.parallelize(range(100)) mapped rdd.map(lambda x: (x % 10, x)) reduced mapped.reduceByKey(lambda a, b: a b) print(reduced.toDebugString())输出里你会看到类似(10) ShuffledRDD[...]的字样ShuffledRDD就是宽依赖的标志。如果是窄依赖会显示MapPartitionsRDD之类的。这个方法适合快速确认某段逻辑有没有触发 shuffle。6.2 用 Spark UI 看 Stage 和 Shuffle 读写量任务跑起来后打开 Spark UI 的 Stages 页面能看到每个 Stage 的划分、任务数、shuffle read 和 shuffle write 的字节数。如果某个 Stage 的 shuffle write 特别大说明宽依赖的数据量很大值得优化。我排查性能问题时第一步永远是看 Spark UIStage 有几个、哪个 Stage 耗时最长、shuffle 读写量多少、有没有明显的长尾任务。这几个信息基本能定位到问题在哪个环节。6.3 用 explain 看物理执行计划DataFrame 和 Dataset API 可以用explain(true)看物理计划里面会标出Exchange节点Exchange就是 shuffle 的物理体现。RDD API 没有 explain但可以通过toDebugString和 UI 结合判断。df.groupBy(key).count().explain(True)输出里出现Exchange hashpartitioning就说明这里有一次宽依赖。如果你发现计划里有多次 Exchange可以考虑能不能合并或消除。7. 常见问题与排查速查表7.1 为什么我的任务 Stage 特别多Stage 多通常意味着 shuffle 多。排查思路先看代码里用了哪些宽依赖算子尤其是groupByKey、join、repartition、distinct这些。然后看能不能用窄依赖替代或者合并多次 shuffle。比如先repartition再sortByKey可以换成repartitionAndSortWithinPartitions一次 shuffle 搞定两件事。7.2 为什么 reduce 阶段特别慢大概率是数据倾斜。去 Spark UI 看各个任务的耗时分布如果少数任务耗时远超其他任务基本可以确认。解决办法按优先级能广播小表就广播不能广播就加盐打散再不行就提高并行度缓解。7.3 coalesce 和 repartition 到底怎么选coalesce不 shuffle适合减少分区数比如写完数据后合并小文件。repartition会 shuffle适合增加分区数或需要均匀分布数据。记住一点coalesce减分区是窄依赖repartition无论增减都是宽依赖。7.4 缓存了 RDD 为什么还是慢可能是缓存级别不对内存不够导致频繁落盘也可能是缓存时机不对缓存在了宽依赖之后每次还是要 shuffle。缓存要放在宽依赖之前、被多次复用的 RDD 上效果才明显。问题现象可能原因排查手段解决方向Stage 数量异常多宽依赖算子过多看 toDebugString 和 UI合并 shuffle、替换算子个别任务耗时极长数据倾斜看 UI 任务耗时分布加盐、广播、提并行度shuffle 读写量巨大groupByKey 等全量 shuffle看 UI shuffle 指标换 reduceByKey、map 端预聚合缓存后仍慢缓存级别或时机不当看 Storage 页面调整级别、前移缓存点join 特别慢分区器不一致触发 shuffle看物理计划 Exchange统一分区器、广播小表7.5 几个我踩过的坑第一个坑以为filter之后数据变少shuffle 就会快。实际上filter是窄依赖它减少的是数据量但不会改变宽依赖的存在。如果filter放在宽依赖之后那 shuffle 的数据量一点没少。正确做法是尽量把filter往前放在 shuffle 之前就把不需要的数据过滤掉。第二个坑用distinct去重时没注意它底层是mapreduceByKeymap中间有一次完整 shuffle。如果数据量很大distinct会很贵。可以考虑用 Bloom Filter 做近似去重或者先按业务规则缩小范围再去重。第三个坑join时两张表都很大又都没分区器结果 shuffle 数据量爆炸。后来改成先用相同分区器partitionBy再 joinshuffle 量直接降了一半多。8. 宽窄依赖的优化思路把能省的 shuffle 都省掉8.1 map 端预聚合reduceByKey 为什么比 groupByKey 香reduceByKey会在 map 端先做一次局部合并相同 key 在同一个 map 分区内的数据先聚一次再发给 reduce 端。groupByKey不做这个预聚合所有原始数据全量 shuffle。假设一个分区里有 100 万条(key, 1)reduceByKey在 map 端可能就聚成几千条groupByKey则是 100 万条全发出去。差距是数量级的。同理aggregateByKey、combineByKey也支持 map 端预聚合适合做更复杂的聚合逻辑。foldByKey也是。只要聚合操作满足结合律就优先用这些带 combine 的算子。8.2 广播变量小表 join 大表的利器如果 join 的两张表一大一小小表能放进内存就用广播变量把小表发到每个 executor大表在 map 端直接查小表做 join完全避开 shuffle。这就是 map-side join。small {k1: v1, k2: v2} bc sc.broadcast(small) big_rdd.map(lambda x: (x, bc.value.get(x[0], None)))广播变量适合小表几百 MB 以内的场景太大广播本身也会成为瓶颈。判断标准是小表大小乘以 executor 数量别超过集群可用内存。8.3 分区器复用让 join 变成窄依赖前面提过两个 RDD 用相同分区器分区后join 可以变成窄依赖。实操上可以在数据第一次进入时就按 join key 分区后续多次 join 都复用这个分区器。代价是第一次分区有一次 shuffle但后续每次 join 都省了。partitioner new HashPartitioner(200) rdd_a rdd_a.partitionBy(partitioner) rdd_b rdd_b.partitionBy(partitioner) joined rdd_a.join(rdd_b) # 窄依赖无 shuffle这个优化在多表关联、迭代计算场景里收益很大。缺点是分区器要提前规划好分区数选得不合适反而会拖慢。8.4 合理设置并行度别让分区数成为瓶颈分区数太少并行度上不去任务跑不满集群分区数太多任务调度开销大小文件多。一般建议每个分区处理 128MB 到 256MB 数据具体看集群配置和任务类型。shuffle 分区数默认是 200数据量大时可以调高数据量小则调低。我通常会在 shuffle 算子后面跟一个coalesce或repartition来调整最终输出分区数避免产生大量小文件。写 HDFS 时尤其要注意小文件多了会影响后续读取性能。9. 我个人在实际操作中的几点体会宽窄依赖这个概念刚学的时候觉得就是个分类背下来应付面试就行。真正做过几个生产任务之后才发现它是理解 Spark 性能的钥匙。你写的每一行代码最终都会映射到 DAG 上的某条边是窄是宽决定了数据要不要搬家、任务要不要等待、失败要不要重算。我现在写 Spark 代码有个习惯写完先不跑先在脑子里过一遍 DAG数一数有几个宽依赖、几次 shuffle、数据量大概多少。如果发现某次 shuffle 完全可以避免就回头改代码。这个习惯帮我省下了大量调优时间。还有一点别迷信“算子替换就能解决一切”。reduceByKey比groupByKey好但前提是你的聚合逻辑能用 reduce 表达。如果业务上确实需要拿到每个 key 的全部 value 列表那groupByKey就是合理选择这时候该考虑的是加内存、调并行度、处理倾斜而不是硬换算子。最后分享一个小技巧如果你不确定某段逻辑会不会触发 shuffle写个最小复现的 demo用toDebugString打印一下比翻文档快得多。Spark 的 API 文档不会告诉你 Stage 怎么切但toDebugString和 Spark UI 会。多动手验证比背概念管用。
