Spark RDD依赖关系详解：窄依赖、宽依赖与Stage划分原理

Spark RDD依赖关系详解：窄依赖、宽依赖与Stage划分原理

发布时间：2026/10/14 3:12:55
Spark RDD依赖关系详解：窄依赖、宽依赖与Stage划分原理 1. RDD依赖关系到底在解决什么问题很多人学Spark的时候前面几个章节都挺顺——RDD是什么、怎么创建、Transformation和Action怎么区分这些概念看一遍就懂了。但一到“依赖关系”这块就开始犯迷糊什么宽依赖窄依赖什么Stage划分DAG又是什么鬼感觉突然从“写代码”跳到了“系统设计”的层面。我刚开始接触这块的时候也有同样的困惑。后来真正在生产环境跑过几百个节点的任务、调过各种数据倾斜和Shuffle爆炸的坑之后才慢慢理解RDD的依赖关系本质上就是Spark整个执行引擎的“骨架”。没有它Spark不可能做到容错恢复不可能做Stage划分更不可能做任务调度优化。打个比方。你把整个Spark作业想象成一条流水线RDD就是流水线上的一个个工位而依赖关系就是工位之间的传送带。传送带怎么连、连了几条、是直连还是需要中转直接决定了整条流水线的效率和出问题时能不能快速恢复。具体来说依赖关系解决了三个核心问题容错恢复某个分区的数据丢了Spark不需要从头重算整个作业只需要根据依赖关系回溯到丢失的那个分区重新计算那一条链路就行。这就像你写代码时用Git改坏了一个文件不需要回滚整个项目只需要checkout那一个文件。Stage划分Spark需要根据依赖关系来决定哪些计算可以串在一起放在同一个Stage里哪些必须等上一步全部完成才能开始需要切分Stage。这直接影响到任务的并行度和执行效率。任务调度优化知道了依赖关系Spark才能做本地性优化、推测执行、动态资源分配这些高级功能。这篇文章我会从实操角度出发把RDD依赖关系这块彻底讲透。不管你是刚学Spark的新手还是已经写过一些作业但没深究过底层原理的老手看完之后应该都能对Spark的执行机制有一个清晰的认识。我会用大量代码示例和实际运行结果来佐证确保每个概念你都能看得见、摸得着。2. 核心概念拆解窄依赖与宽依赖2.1 从一次简单的map操作说起先看一段最基础的代码from pyspark import SparkConf, SparkContext conf SparkConf().setMaster(local[*]).setAppName(DependencyDemo) sc SparkContext(confconf) rdd1 sc.parallelize([1, 2, 3, 4, 5], 3) rdd2 rdd1.map(lambda x: x * 2) rdd3 rdd2.filter(lambda x: x 4) print(rdd3.collect()) # 输出: [6, 8, 10]这段代码里rdd2依赖rdd1rdd3依赖rdd2。每个分区的计算都是独立的——rdd1的3个分区各自做map变成rdd2的3个分区rdd2的3个分区各自做filter变成rdd3的3个分区。父RDD的每个分区最多只被子RDD的一个分区使用这就是窄依赖。窄依赖的核心特征分区之间的对应关系是确定的、一对一的或者多对一的比如union。计算某个子分区时不需要等待其他分区的数据。2.2 宽依赖是怎么产生的再看另一段代码rdd1 sc.parallelize([(a, 1), (b, 2), (a, 3), (c, 4), (b, 5)], 3) rdd2 rdd1.reduceByKey(lambda x, y: x y) print(rdd2.collect()) # 输出: [(a, 4), (b, 7), (c, 4)]这里reduceByKey会产生宽依赖。为什么因为要按key做聚合相同key的数据必须被拉到同一个分区里。rdd1的3个分区中key为a的数据可能分布在不同的分区上要完成聚合就必须把数据重新分发——这个过程就是Shuffle。宽依赖的核心特征父RDD的每个分区可能被多个子RDD分区使用也就是说子RDD的一个分区通常需要等待父RDD所有分区中属于它的那部分数据都准备好。这就引入了一个“等待屏障”Spark必须等上一个Stage全部完成才能进入下一个Stage。2.3 一张表说清楚两者的区别对比维度窄依赖宽依赖分区对应关系一对一或多对一一对多或多对多是否产生Shuffle否是能否在同一个Stage内可以不可以必须切分Stage容错恢复代价低只需重算对应分区高需要重算整个父RDD典型算子map、filter、flatMap、uniongroupByKey、reduceByKey、join数据传输不需要跨节点需要跨节点网络传输注意join算子比较特殊。如果两个RDD在join之前已经按照相同的分区器进行了分区那么join可能是窄依赖否则就是宽依赖。这个细节在实际调优中非常关键。2.4 为什么窄依赖的容错代价更低假设rdd3的partition 1丢了。如果是窄依赖Spark只需要找到rdd2的partition 1重新执行一次filter就行。但如果rdd3是通过宽依赖从rdd2来的那partition 1的数据可能来自rdd2的所有分区Spark就必须把rdd2的所有分区都重算一遍然后重新Shuffle。这就是为什么Spark在设计上尽量鼓励使用窄依赖的算子。比如用reduceByKey而不是groupByKey前者会在map端做预聚合combiner减少Shuffle的数据量后者则会把所有数据原封不动地Shuffle一遍。3. DAG与Stage划分的底层逻辑3.1 DAG是什么为什么需要它DAG全称是Directed Acyclic Graph有向无环图。Spark把用户写的所有Transformation操作记录下来形成一个DAG。这个DAG的节点是RDD边是依赖关系。为什么是“有向无环”因为RDD的转换是单向的——从父RDD到子RDD不可能出现循环依赖。这个特性保证了Spark可以按照拓扑顺序依次执行不会死循环。当你调用一个Action算子比如collect、count、saveAsTextFile时Spark才会真正触发计算。这时候它会做几件事从最终RDD开始反向遍历依赖链构建DAG根据依赖类型窄依赖还是宽依赖划分Stage为每个Stage生成TaskSet提交给TaskSchedulerTaskScheduler把Task分发到Executor上执行3.2 Stage划分的具体规则Stage划分的规则其实很简单从后往前推遇到宽依赖就切一刀。具体来说Spark从触发Action的那个RDD开始反向遍历依赖链。如果遇到窄依赖就把当前RDD和父RDD归到同一个Stage如果遇到宽依赖就以宽依赖为边界切分成两个Stage。rdd1 sc.parallelize(range(100), 4) rdd2 rdd1.map(lambda x: (x % 10, x)) # 窄依赖 rdd3 rdd2.reduceByKey(lambda a, b: a b) # 宽依赖切分点 rdd4 rdd3.filter(lambda x: x[1] 50) # 窄依赖 rdd5 rdd4.mapValues(lambda x: x * 2) # 窄依赖 result rdd5.collect()这段代码会被划分成两个StageStage 0包含rdd1 - rdd2的计算以及reduceByKey的Shuffle Write部分Stage 1包含reduceByKey的Shuffle Read部分以及rdd3 - rdd4 - rdd5的计算Stage 0先执行把Shuffle数据写到磁盘Stage 1再执行从磁盘读取Shuffle数据继续计算。3.3 为什么要这样切分核心原因在于宽依赖需要等待所有父分区数据就绪。如果reduceByKey和后面的filter放在同一个Stage里那filter的Task就必须等所有reduceByKey的Task都完成才能开始这会导致大量Task空等资源利用率极低。切分成两个Stage之后Stage 0的所有Task可以并行执行完成后Stage 1的Task再并行执行。虽然多了一次磁盘I/OShuffle Write和Read但换来了更好的并行度和更清晰的执行边界。实操心得在Spark UI上可以看到每个Stage的Task数量、执行时间、Shuffle数据量。如果你发现某个Stage的Task数量特别少比如只有1个那大概率是上游用了repartition(1)或者coalesce(1)这会导致严重的性能瓶颈。3.4 一个容易搞混的点Stage数量不等于Task数量很多人以为一个Stage就是一个Task这是不对的。一个Stage包含的Task数量等于该Stage最后一个RDD的分区数。比如Stage 0的最后一个RDD是rdd2它有4个分区那Stage 0就有4个Task。Stage 1的Task数量取决于reduceByKey之后的分区数。默认情况下reduceByKey使用父RDD的分区数也就是4。但你可以通过第二个参数指定rdd3 rdd2.reduceByKey(lambda a, b: a b, 8) # 指定8个分区这样Stage 1就会有8个Task。4. 实操验证用代码亲眼看到依赖关系4.1 查看RDD的依赖类型Spark提供了dependencies方法可以直接查看一个RDD的依赖rdd1 sc.parallelize([1, 2, 3, 4, 5], 3) rdd2 rdd1.map(lambda x: x * 2) print(rdd2.dependencies()) # 输出: [pyspark.rdd.PipelinedRDD object at ...] # PipelinedRDD 表示窄依赖 rdd3 rdd2.map(lambda x: (x % 2, x)) rdd4 rdd3.reduceByKey(lambda a, b: a b) print(rdd4.dependencies()) # 输出: [pyspark.rdd.ShuffledRDD object at ...] # ShuffledRDD 表示宽依赖在PySpark中窄依赖通常表现为PipelinedRDD宽依赖表现为ShuffledRDD。这个判断方法虽然不够精确有些窄依赖也会产生其他类型的RDD但在大多数场景下够用了。4.2 用toDebugString查看血缘关系toDebugString()方法可以打印出RDD的完整血缘链rdd1 sc.parallelize([1, 2, 3, 4, 5], 3) rdd2 rdd1.map(lambda x: x * 2) rdd3 rdd2.filter(lambda x: x 4) rdd4 rdd3.map(lambda x: (x % 3, x)) rdd5 rdd4.reduceByKey(lambda a, b: a b) print(rdd5.toDebugString().decode(utf-8))输出大概长这样(3) PythonRDD[5] at RDD at PythonRDD.scala:53 [] | PythonRDD[4] at RDD at PythonRDD.scala:53 [] | PythonRDD[3] at RDD at PythonRDD.scala:53 [] | PythonRDD[2] at RDD at PythonRDD.scala:53 [] | ParallelCollectionRDD[1] at parallelize at PythonRDD.scala:195 []注意看缩进——每一层缩进代表一次依赖。如果中间有Shuffle输出中会出现ShuffledRDD的标记并且缩进会发生变化。4.3 在Spark UI上观察Stage划分启动Spark作业后打开Spark UI默认端口4040进入“Jobs”页面点击具体的Job就能看到Stage的划分情况。每个Stage会显示Stage ID和名称包含的Task数量Shuffle Read/Write的数据量执行时间我实测过一个稍微复杂一点的作业在Spark UI上看到的Stage划分和代码里的宽依赖位置完全对应。这个验证过程对理解Stage划分非常有帮助建议你也动手试一下。注意事项在local模式下运行时Spark UI的端口是4040。如果4040被占用了会自动切换到4041、4042等。你可以在启动日志里找到实际端口。4.4 一个完整的实验对比窄依赖和宽依赖的执行差异我写了一个对比实验分别用窄依赖和宽依赖完成同样的聚合逻辑观察执行时间和Shuffle数据量import time # 准备数据100万条记录10个分区 data [(i % 100, i) for i in range(1000000)] rdd sc.parallelize(data, 10) # 方式一使用reduceByKey宽依赖但有map端预聚合 start time.time() result1 rdd.reduceByKey(lambda a, b: a b).collect() time1 time.time() - start print(freduceByKey耗时: {time1:.3f}秒) # 方式二使用groupByKey宽依赖无预聚合 start time.time() result2 rdd.groupByKey().mapValues(sum).collect() time2 time.time() - start print(fgroupByKey耗时: {time2:.3f}秒)在我本地环境4核8G实测下来reduceByKey比groupByKey快了将近40%。原因很简单reduceByKey在map端先做了一次本地聚合Shuffle的数据量大大减少而groupByKey把所有原始数据都Shuffle了一遍。这个实验说明了一个重要道理同样是宽依赖不同算子的性能差异可能非常大。理解依赖关系的类型只是第一步更重要的是知道怎么根据业务场景选择最合适的算子。5. 常见问题与排查技巧实录5.1 Shuffle数据量过大导致OOM这是生产环境最常见的问题之一。表现是某个Stage的Shuffle Write数据量特别大Executor频繁GC甚至OOM。排查思路在Spark UI的Stage页面查看Shuffle Write和Shuffle Read的数据量如果Shuffle Write远大于输入数据量说明map端没有做足够的预聚合检查是否用了groupByKey而不是reduceByKey或aggregateByKey检查是否有大量key倾斜——某个key的数据量远超其他key解决办法用reduceByKey替代groupByKey对key加盐加随机前缀打散热点key增加Shuffle分区数spark.sql.shuffle.partitions默认200或reduceByKey(func, numPartitions)5.2 Stage执行时间过长Task数量太少如果某个Stage只有1个Task那不管集群有多少资源这个Stage都只能在一个Executor上跑。这通常是因为上游用了repartition(1)或coalesce(1)。排查方法在Spark UI上查看Stage的Task数量。如果Task数量远小于集群的CPU核心数说明并行度不足。解决办法调整分区数。一般来说每个CPU核心处理2-4个分区比较合适。比如集群有100个核心那分区数设置在200-400之间比较合理。5.3 宽依赖导致的重复计算这个问题比较隐蔽。如果你的DAG中有多个Action算子共享同一个上游RDD而且中间有宽依赖那每次Action都会触发一次完整的Shuffle。rdd1 sc.parallelize(data, 10) rdd2 rdd1.map(lambda x: (x % 10, x)) rdd3 rdd2.reduceByKey(lambda a, b: a b) # 第一次Action触发Shuffle count rdd3.count() # 第二次Action又触发一次Shuffle total rdd3.mapValues(lambda x: x).reduce(lambda a, b: (a[0], a[1] b[1]))解决办法对rdd3调用cache()或persist()把Shuffle后的结果缓存起来。rdd3.cache() count rdd3.count() # 第一次触发Shuffle并缓存 total rdd3.mapValues(...).reduce(...) # 直接从缓存读取实操心得cache()默认使用MEMORY_ONLY级别如果数据量太大放不下会丢弃部分分区下次用到时重新计算。生产环境建议根据数据量选择MEMORY_AND_DISK或MEMORY_AND_DISK_SER。5.4 常见问题速查表问题现象可能原因排查方法解决方案Shuffle数据量过大map端无预聚合查看Spark UI的Shuffle Write用reduceByKey替代groupByKey某个Task特别慢数据倾斜查看Task的Duration分布加盐打散热点keyStage只有1个Task分区数太少查看Stage的Task数量repartition增加分区重复Shuffle未缓存中间结果查看DAG中是否有重复的Shuffle对中间RDD调用cacheExecutor OOM单分区数据量过大查看GC日志和Shuffle数据量增加分区数或提高Executor内存5.5 一个真实踩坑案例之前有个作业逻辑是从Hive表读数据做一系列map/filter然后reduceByKey聚合最后写入HDFS。代码看起来没问题但跑起来特别慢一个Stage跑了将近2小时。排查过程打开Spark UI发现Shuffle Write数据量是输入数据的3倍——明显不正常检查代码发现reduceByKey之前有一个flatMap操作把每条记录展开成了多条更关键的是flatMap之后没有做filter导致大量无用数据参与了Shuffle在flatMap之后加了一个filterShuffle数据量直接降到了原来的1/5执行时间从2小时缩短到20分钟这个案例的教训是宽依赖之前的操作要特别小心任何数据膨胀都会在Shuffle阶段被放大。能在窄依赖阶段过滤掉的数据千万不要留到宽依赖之后。6. 依赖关系对性能调优的实际影响6.1 算子选择窄依赖优先原则在能满足业务需求的前提下优先选择窄依赖的算子。比如用mapPartitions替代map减少函数调用开销用filter尽早过滤数据减少后续Shuffle的数据量用reduceByKey替代groupByKeymap端预聚合用aggregateByKey替代groupByKey更灵活的聚合这些选择看起来是算子层面的优化但本质上都是在减少宽依赖带来的Shuffle开销。6.2 分区器复用避免不必要的Shuffle如果两个RDD需要多次join而且join的key相同可以先把它们用相同的分区器分区这样后续的join就不需要再Shuffle了。from pyspark import SparkContext rdd1 sc.parallelize([(a, 1), (b, 2), (c, 3)]) rdd2 sc.parallelize([(a, 4), (b, 5), (d, 6)]) # 使用HashPartitioner分区数为4 rdd1_partitioned rdd1.partitionBy(4) rdd2_partitioned rdd2.partitionBy(4) # 第一次join触发Shuffle result1 rdd1_partitioned.join(rdd2_partitioned) # 第二次join如果还是同样的分区器不会触发Shuffle rdd3 sc.parallelize([(a, 7), (b, 8), (c, 9)]).partitionBy(4) result2 rdd1_partitioned.join(rdd3)这个技巧在需要多次join的场景下特别有用可以显著减少Shuffle次数。6.3 广播变量把小表放进内存如果join的一方是小表比如几MB的维度表可以用广播变量把小表分发到每个Executor这样join就变成了窄依赖完全避免了Shuffle。small_table {a: 1, b: 2, c: 3} broadcast_var sc.broadcast(small_table) rdd sc.parallelize([(a, 100), (b, 200), (d, 300)]) result rdd.map(lambda x: (x[0], x[1] broadcast_var.value.get(x[0], 0)))注意事项广播变量的大小默认限制是8GBspark.sql.autoBroadcastJoinThreshold但实际使用中建议控制在几百MB以内否则网络传输和内存开销会很大。6.4 依赖关系与数据本地性窄依赖的一个隐藏优势是数据本地性更好。因为窄依赖不需要跨节点传输数据Task可以在数据所在的节点上直接执行避免了网络I/O。宽依赖则不可避免地需要网络传输。Spark会尽量把Shuffle Read的Task调度到Shuffle Write数据所在的节点上但这取决于集群的资源和调度策略不一定总能做到。在实际调优中如果发现某个Stage的网络I/O特别高可以考虑增加Executor的内存让更多Shuffle数据留在内存中调整spark.reducer.maxSizeInFlight参数控制每次拉取的数据量使用SSD作为Shuffle的临时存储目录7. 从依赖关系看Spark的设计哲学7.1 惰性求值与依赖链Spark的惰性求值机制和依赖关系是紧密配合的。Transformation算子不会立即执行只是记录依赖关系Action算子触发执行时Spark才根据依赖链生成物理执行计划。这种设计的好处是Spark可以在执行前对依赖链做全局优化。比如它可以把多个窄依赖合并成一个Stage减少中间结果的物化也可以在宽依赖处做Shuffle优化比如合并多个Shuffle操作。7.2 容错机制血缘关系的价值RDD的血缘关系Lineage是Spark容错的核心。每个RDD都记录了它是从哪些父RDD经过什么操作计算出来的。当某个分区丢失时Spark可以沿着血缘关系回溯重新计算丢失的分区。这种容错机制比传统的副本机制更轻量——不需要为每个分区维护多个副本只需要记录计算逻辑。代价是恢复时需要重新计算但对于大多数场景来说这个代价是可以接受的。7.3 与MapReduce的对比MapReduce只有Map和Reduce两个阶段中间必须经过一次Shuffle。Spark的DAG模型允许更灵活的Stage划分可以把多个窄依赖的操作串在一起减少了不必要的磁盘I/O和网络传输。这也是Spark比MapReduce快的重要原因之一。在MapReduce中每个MapReduce作业都会把中间结果写到HDFS而Spark可以把多个窄依赖操作放在一个Stage里中间结果直接在内存中传递。7.4 依赖关系对后续学习的影响理解了依赖关系之后再学Spark的其他高级特性就会顺畅很多Spark SQLCatalyst优化器会根据依赖关系做谓词下推、列裁剪等优化Structured Streaming流处理的微批次本质上也是基于依赖关系的DAGMLlib机器学习算法的迭代计算需要反复利用缓存的RDD依赖关系决定了缓存的粒度可以说依赖关系是理解Spark整个技术栈的基础。这块搞清楚了后面学什么都快。8. 几个容易被忽略的细节8.1 checkpoint与依赖关系checkpoint会把RDD的数据持久化到可靠存储通常是HDFS同时切断血缘关系。这在长血缘链的场景下非常有用——可以避免恢复时重算整个链条。sc.setCheckpointDir(hdfs://path/to/checkpoint) rdd.checkpoint()注意checkpoint是惰性的需要触发一次Action才会真正执行。而且checkpoint之前最好先cache()否则会触发两次计算。8.2 依赖关系与序列化宽依赖涉及Shuffle需要对数据进行序列化。Spark支持多种序列化方式默认是Java序列化但推荐使用Kryo序列化性能更好。conf SparkConf().set(spark.serializer, org.apache.spark.serializer.KryoSerializer)Kryo序列化的体积通常比Java序列化小几倍能显著减少Shuffle的网络传输量。8.3 依赖关系与推测执行推测执行Speculative Execution是Spark的一种优化机制如果某个Task执行特别慢Spark会在另一个Executor上启动一个相同的Task谁先完成就用谁的结果。这个机制对宽依赖的Stage特别有用因为宽依赖的Task通常执行时间更长更容易出现个别Task拖后腿的情况。可以通过spark.speculation参数开启。8.4 依赖关系与动态资源分配动态资源分配允许Spark根据负载自动增减Executor。在宽依赖的Stage中因为需要等待所有Task完成资源需求会有明显的波峰波谷。动态资源分配可以在波谷时释放资源波峰时申请资源提高集群利用率。配置方法conf.set(spark.dynamicAllocation.enabled, true) conf.set(spark.dynamicAllocation.minExecutors, 2) conf.set(spark.dynamicAllocation.maxExecutors, 50)我在实际使用中发现动态资源分配在共享集群上特别有用可以避免一个作业占着大量资源不干活的情况。但要注意设置合理的min和max避免频繁申请释放导致调度开销过大。8.5 一个关于cache的小技巧很多人知道要cache中间结果但不知道cache的时机也很重要。最佳实践是在第一次触发宽依赖的Action之前cache。rdd1 sc.parallelize(data, 10) rdd2 rdd1.map(lambda x: (x % 10, x)) rdd3 rdd2.reduceByKey(lambda a, b: a b) # 在Action之前cache rdd3.cache() # 第一次Action触发Shuffle并缓存 rdd3.count() # 后续Action直接从缓存读取 rdd3.filter(lambda x: x[1] 100).count()如果cache的时机太晚比如在第一次Action之后那第一次Action的Shuffle结果不会被缓存后续Action还是需要重新计算。8.6 依赖关系与数据倾斜的深层联系数据倾斜的本质是某些key的数据量远大于其他key导致对应的Task执行时间特别长。在宽依赖的Stage中这个问题会被放大——因为所有Task都必须完成才能进入下一个Stage一个慢Task会拖慢整个Stage。解决数据倾斜的方法有很多但核心思路都是打散热点key。比如# 对key加随机前缀 rdd2 rdd1.map(lambda x: (str(random.randint(0, 9)) _ x[0], x[1])) rdd3 rdd2.reduceByKey(lambda a, b: a b) # 去掉前缀后再聚合一次 rdd4 rdd3.map(lambda x: (x[0].split(_)[1], x[1])).reduceByKey(lambda a, b: a b)这个方法虽然多了一次Shuffle但能有效解决热点key问题。具体加多少个前缀取决于热点key的数据量和集群的并行度。9. 从依赖关系出发的调优清单基于我对依赖关系的理解整理了一份调优清单供你在实际项目中参考调优项具体操作预期效果算子选择用reduceByKey替代groupByKey减少Shuffle数据量50%以上分区数调整设置合理的Shuffle分区数提高并行度避免Task过少广播变量小表join用广播消除Shuffle变成窄依赖分区器复用多次join用相同分区器减少重复Shuffle中间结果缓存对宽依赖后的RDD调用cache避免重复计算序列化优化使用Kryo序列化减少网络传输量数据倾斜处理加盐打散热点key解决个别Task过慢问题checkpoint长血缘链设置checkpoint加速容错恢复这份清单不是万能的具体怎么调还要看业务场景和数据特征。但只要你理解了依赖关系的本质就能根据实际情况灵活运用这些技巧。我在实际项目中最大的体会是不要等到作业跑不动了才去调优。在写代码的时候就应该有意识地去设计依赖关系——哪些地方会产生Shuffle、Shuffle的数据量大概有多大、有没有办法减少Shuffle。这种前置思考比事后调优有效得多。另外Spark UI是你最好的朋友。每次写完一个作业花几分钟看一下Stage划分、Shuffle数据量、Task执行时间分布长期积累下来你对依赖关系的直觉会越来越准。