1. 为什么RDD依赖关系是Spark调度的命门很多人学Spark前面几个章节把RDD的基本概念、创建方式、常见算子都过了一遍感觉已经能写代码跑任务了结果一到调优或者排查性能问题就懵了。任务为什么被拆成这么多Stage为什么有些操作会触发Shuffle而有些不会为什么某个任务卡在某个阶段迟迟不动这些问题的答案全都藏在RDD的依赖关系里。说白了RDD的依赖关系就是Spark整个执行引擎的“骨架”。你写的每一行Transformation操作Spark并不会立刻执行而是在后台默默构建一张有向无环图DAG这张图上的边就是依赖关系。等到遇到Action算子时Spark才会根据这些依赖关系把DAG切分成多个Stage然后逐个提交执行。如果你不理解依赖关系你就无法理解Spark为什么要这样切Stage也就无法针对性地做性能优化。这篇文章面向的是已经接触过Spark基础、能写简单RDD操作但还没深入理解调度原理的开发者。我会从依赖关系的本质出发把宽依赖和窄依赖的区别讲透然后带你搞清楚Stage是怎么切出来的最后分享几个我在实际项目中因为依赖关系踩过的坑和调优经验。整篇内容基于Spark Core的RDD模型展开不涉及DataFrame或Dataset的Catalyst优化器那套东西那些后面单独聊。注意本文讨论的依赖关系特指RDD之间的血缘关系Lineage不是指任务之间的调度依赖也不是指外部数据源的依赖。这几个概念容易混淆初学时一定要区分清楚。2. 窄依赖与宽依赖的本质区别到底在哪2.1 从“一个子RDD分区依赖几个父RDD分区”说起判断两个RDD之间是窄依赖还是宽依赖最核心的标准就一句话子RDD的每个分区是否只依赖于父RDD的固定数量分区。如果子RDD的每个分区只依赖父RDD的一个或少数几个固定分区那就是窄依赖如果子RDD的每个分区可能依赖父RDD的所有分区那就是宽依赖。这个定义听起来简单但很多人第一次看的时候会犯迷糊。我换个说法你就明白了窄依赖就像一条流水线每个工位只从上一个工位拿固定数量的半成品不需要等其他工位的货宽依赖就像开大会每个小组要汇总所有人的数据才能出结果必须等所有人都到齐。常见的窄依赖操作包括map、filter、flatMap、union、co-partitioned join等。这些操作的特点是父RDD的每个分区最多被子RDD的一个分区使用数据不需要跨节点传输。常见的宽依赖操作包括groupByKey、reduceByKey、sortByKey、join非co-partitioned、repartition、distinct等。这些操作的特点是子RDD的每个分区可能依赖父RDD的多个甚至全部分区数据需要经过网络Shuffle。2.2 为什么Spark要区分这两种依赖这个问题很关键理解了它你就理解了Spark调度设计的精髓。窄依赖之所以“窄”是因为它允许Spark做流水线Pipeline执行。什么意思呢就是父RDD的每个分区计算出结果后可以直接在同一个Task里继续执行子RDD的转换不需要落盘也不需要等其他分区。这样整个链条上的多个窄依赖操作可以合并成一个Stage在一个Task里一口气跑完。宽依赖就不行了。因为子RDD的每个分区需要父RDD的所有分区数据Spark必须等父RDD所有分区都计算完毕然后把数据按照key重新分发到不同节点上这个过程就是Shuffle。Shuffle是Spark里最昂贵的操作涉及磁盘I/O、网络传输、序列化反序列化所以Spark必须把宽依赖作为Stage的边界在Shuffle前后各切一个Stage。我画个简单的对比表帮你记忆对比维度窄依赖宽依赖分区依赖关系子RDD每个分区依赖父RDD固定数量分区子RDD每个分区依赖父RDD所有分区是否触发Shuffle否是能否Pipeline执行能多个窄依赖可合并为一个Stage不能宽依赖是Stage边界数据是否需要跨节点传输不需要需要失败恢复代价低只需重算对应分区高需要重算所有父分区典型算子map、filter、flatMap、uniongroupByKey、reduceByKey、join2.3 一个容易搞混的案例join到底是宽依赖还是窄依赖很多人以为join一定是宽依赖其实不一定。如果两个RDD在join之前已经按照相同的分区器Partitioner进行了分区并且分区数量一致那么Spark可以做到co-partitioned join这时候每个子RDD分区只需要从对应的父RDD分区拉取数据不需要全量Shuffle这就是窄依赖。但如果你没有提前做好分区直接对两个普通RDD做joinSpark默认会走Shuffle这就是宽依赖。所以我在实际项目中经常做的一件事就是如果两个RDD要反复join我会提前用partitionBy把它们按相同的key分区这样后续的join就变成了窄依赖性能提升非常明显。# 假设rdd1和rdd2都是(key, value)格式 # 提前按相同分区器分区后续join变成窄依赖 partitioner rdd1.partitioner or rdd2.partitioner if partitioner: rdd1_partitioned rdd1.partitionBy(partitioner) rdd2_partitioned rdd2.partitionBy(partitioner) joined rdd1_partitioned.join(rdd2_partitioned) # 窄依赖提示判断join是否为窄依赖关键看两个RDD是否已经co-partitioned。如果没有Spark会在join时自动触发Shuffle这时候就是宽依赖。3. DAG如何根据依赖关系切分Stage3.1 从Action算子反向推导Stage边界Spark切分Stage的规则其实很朴素从后往前推遇到宽依赖就切一刀。具体来说当你触发一个Action算子时Spark会从该Action所在的RDD开始沿着依赖关系向前回溯。如果遇到窄依赖就继续往前如果遇到宽依赖就在宽依赖处切分宽依赖之前的所有窄依赖操作归为一个Stage宽依赖之后的操作归为另一个Stage。举个例子假设你有这样一条链rdd sc.textFile(data.txt) rdd1 rdd.filter(lambda x: error in x) rdd2 rdd1.map(lambda x: (x.split(,)[0], 1)) rdd3 rdd2.reduceByKey(lambda a, b: a b) rdd4 rdd3.map(lambda x: (x[1], x[0])) rdd4.collect()从collect往前推rdd4依赖rdd3是窄依赖map继续往前rdd3依赖rdd2是宽依赖reduceByKey在这里切一刀。所以rdd、rdd1、rdd2归为一个StageStage 0rdd3、rdd4归为另一个StageStage 1。Stage 0负责读取数据并做filter和map然后Shuffle输出Stage 1负责接收Shuffle数据做reduceByKey和map最后collect。3.2 Stage类型ShuffleMapStage与ResultStageSpark里有两种StageShuffleMapStage和ResultStage。ShuffleMapStage是中间阶段的Stage它的输出会作为下一个Stage的输入需要把数据写到Shuffle文件中。ResultStage是最后一个Stage它的输出直接对应Action算子的结果不需要再Shuffle。在上面的例子里Stage 0就是ShuffleMapStage因为它后面跟着reduceByKey这个宽依赖需要把map后的数据按key重新分区写盘。Stage 1是ResultStage因为collect是Action直接出结果。一个Job可能包含多个ShuffleMapStage和一个ResultStage。如果DAG中有多个宽依赖就会切出多个ShuffleMapStage。比如你先reduceByKey再join再reduceByKey那就会有三个Stage两个ShuffleMapStage和一个ResultStage。3.3 Stage内的Task数量由什么决定每个Stage内的Task数量等于该Stage最后一个RDD的分区数。比如Stage 0的最后一个RDD是rdd2如果rdd2有10个分区那Stage 0就会启动10个Task并行执行。Stage 1的最后一个RDD是rdd4如果rdd4有5个分区那Stage 1就启动5个Task。这里有个常见的坑如果你在宽依赖之前用了repartition或者coalesce改变了分区数那Stage的Task数量会跟着变。repartition是宽依赖会触发Shufflecoalesce如果不开启shuffle就是窄依赖只是合并分区不触发Shuffle。我在实际项目中经常用coalesce来减少小文件输出但要注意它可能导致某些分区数据量过大需要配合repartition使用。# coalesce不触发Shuffle但可能造成数据倾斜 rdd_coalesced rdd.coalesce(1) # 所有数据合并到1个分区慎用 # repartition触发Shuffle可以均匀分布数据 rdd_repartitioned rdd.repartition(100) # 重新分成100个分区注意coalesce默认不触发Shuffle如果从1000个分区合并到10个分区那10个Task每个要处理100个分区的数据容易OOM。如果数据量大建议用repartition虽然多一次Shuffle但数据分布更均匀。4. 依赖关系在容错恢复中的实际作用4.1 LineageSpark的“后悔药”机制RDD的依赖关系链在Spark里叫Lineage血缘。它的核心作用是容错如果某个分区的数据丢失了比如节点宕机Spark不需要从外部数据源重新读取全部数据而是根据Lineage找到丢失分区的父分区重新计算那部分数据即可。窄依赖的恢复代价很低因为只需要重算对应的父分区。比如map操作丢了分区P1只需要重算父RDD的P1分区即可。但宽依赖的恢复代价就高了因为子RDD的每个分区依赖父RDD的所有分区如果丢了子RDD的一个分区可能需要重算父RDD的所有分区。这就是为什么Spark更倾向于窄依赖也是为什么宽依赖之后的数据最好做Checkpoint。4.2 Checkpoint与依赖链截断当一个RDD的Lineage太长时容错恢复的代价会变得很高。比如你做了100次map每次都是窄依赖虽然不会Shuffle但一旦某个分区丢失需要从源头重新执行100次map。这时候就可以用Checkpoint把中间结果持久化到可靠存储如HDFS同时截断Lineage。Checkpoint之后RDD的依赖关系会被切断后续如果发生故障直接从Checkpoint的数据恢复不需要再往前回溯。我在处理迭代计算如机器学习算法时经常用Checkpoint每迭代10次做一次防止Lineage过长导致恢复时间爆炸。# 设置Checkpoint目录 sc.setCheckpointDir(hdfs://namenode:8020/checkpoint) # 对RDD做Checkpoint rdd.checkpoint() # 触发Action后Checkpoint才会真正执行 rdd.count()4.3 缓存Cache/Persist与依赖关系的关系Cache和Checkpoint不同Cache不会截断Lineage它只是把RDD的数据缓存在内存或磁盘中后续复用该RDD时不需要重新计算。但Cache的RDD如果丢失了仍然需要根据Lineage重新计算。所以Cache适合复用次数多的场景Checkpoint适合Lineage过长或宽依赖后需要截断的场景。我通常的做法是如果一个RDD会被多个Action复用先Cache如果Lineage超过20步或者有宽依赖再考虑Checkpoint。两者可以结合使用先Cache再Checkpoint这样第一次计算后既缓存了数据又截断了Lineage。5. 依赖关系引发的性能问题与调优实战5.1 宽依赖过多导致Stage爆炸我见过一个项目代码里连续写了5个groupByKey每个都触发一次Shuffle结果一个Job切出了6个Stage每个Stage都要等前一个Stage完全写完Shuffle文件才能开始。整个任务跑了40分钟其中30分钟花在Shuffle的磁盘I/O和网络传输上。后来我帮他们改成用reduceByKey替代groupByKey并且在map端做预聚合combineShuffle数据量直接降了70%。reduceByKey会在map端先做本地聚合再发送到reduce端而groupByKey会把所有数据原封不动地Shuffle出去。这个区别在数据量大时非常致命。# 不推荐groupByKey会把所有数据Shuffle出去 rdd.groupByKey().mapValues(sum) # 推荐reduceByKey会在map端预聚合 rdd.reduceByKey(lambda a, b: a b)5.2 数据倾斜宽依赖的“阿喀琉斯之踵”宽依赖最怕的就是数据倾斜。如果某个key的数据量特别大Shuffle时这个key对应的reduce分区就会成为瓶颈其他分区早就跑完了它还在慢慢处理。我在一个日志分析项目里遇到过某个IP的日志量占了总数据的30%reduceByKey时这个IP对应的Task跑了2小时其他Task10分钟就结束了。解决数据倾斜的常用方法有几种一是加盐Salting给热点key加上随机前缀打散到多个分区聚合后再去掉前缀二次聚合二是用repartitionAndSortWithinPartitions自定义分区器把热点key单独处理三是如果热点key可以接受近似结果用sample采样后单独处理。# 加盐处理数据倾斜 import random def add_salt(key): return (key, random.randint(0, 9)) rdd_salted rdd.map(lambda x: (add_salt(x[0]), x[1])) rdd_aggregated rdd_salted.reduceByKey(lambda a, b: a b) rdd_final rdd_aggregated.map(lambda x: (x[0][0], x[1])).reduceByKey(lambda a, b: a b)5.3 合理设置分区数不多不少刚刚好分区数太少Task并行度不够集群资源浪费分区数太多Task启动和调度开销大Shuffle文件数量爆炸。我的经验是分区数设置为集群CPU核心数的2到3倍比较合适。比如集群有100个核心分区数设200到300。但这不是绝对的。如果数据量特别大单个分区数据量超过128MB就需要增加分区数如果数据量很小分区数太多反而慢。我通常先用rdd.getNumPartitions()看看当前分区数再用rdd.mapPartitionsWithIndex统计每个分区的数据量根据实际情况调整。# 查看分区数 print(rdd.getNumPartitions()) # 统计每个分区的数据量 def count_partition(index, iterator): count sum(1 for _ in iterator) yield (index, count) partition_counts rdd.mapPartitionsWithIndex(count_partition).collect() for idx, cnt in partition_counts: print(fPartition {idx}: {cnt} records)5.4 用依赖关系图定位性能瓶颈Spark UI里的DAG可视化是排查性能问题的利器。你可以清楚地看到每个Stage的耗时、Shuffle读写数据量、Task的执行时间分布。如果某个Stage的Shuffle Write特别大说明宽依赖之前的数据量没降下来如果某个Stage的Task执行时间差异很大说明有数据倾斜。我一般会重点关注三个指标Shuffle Read Size、Shuffle Write Size和Task Duration的Max/Median比值。如果Max/Median超过3倍基本可以确定有倾斜如果Shuffle Size超过10GB就要考虑优化宽依赖操作了。6. 几个我在实际项目中踩过的依赖关系坑6.1 误以为filter之后分区数会变filter是窄依赖它不会改变分区数。父RDD有100个分区filter之后还是100个分区只是某些分区的数据可能变少了甚至为空。如果你filter之后直接做reduceByKeyShuffle时会有大量空分区参与浪费资源。我通常会在filter之后加一个coalesce来合并空分区但要注意不要合并得太狠导致数据倾斜。6.2 union之后分区数翻倍但没意识到union是窄依赖它会把两个RDD的分区直接合并分区数等于两个RDD分区数之和。如果你union了10个RDD每个RDD有100个分区那结果就是1000个分区。后续如果做宽依赖操作Shuffle的Task数量就是1000调度开销很大。我一般会在union之后根据数据量做一次repartition或coalesce。6.3 在宽依赖之前没有做map端预聚合前面提过reduceByKey会在map端预聚合但很多人不知道combineByKey也可以而且更灵活。如果你需要自定义聚合逻辑用combineByKey比groupByKey再map好得多。我在一个用户行为分析项目里用combineByKey替代groupByKey后Shuffle数据量从2TB降到了200GB任务时间从3小时缩短到25分钟。# 用combineByKey做map端预聚合 def create_combiner(value): return [value] def merge_value(combiner, value): combiner.append(value) return combiner def merge_combiners(c1, c2): return c1 c2 combined rdd.combineByKey(create_combiner, merge_value, merge_combiners)6.4 忽略了partitionBy的复用价值如果你有多个操作需要按同一个key做Shuffle提前用partitionBy按该key分区后续操作就可以复用这个分区结果避免重复Shuffle。比如你先reduceByKey再join再reduceByKey如果key相同第一次reduceByKey之后就已经按key分区了后续join和reduceByKey都可以利用这个分区变成窄依赖。# 提前按key分区后续操作复用 partitioner rdd.partitioner or None if not partitioner: rdd rdd.partitionBy(100) # 按key分成100个分区 # 后续reduceByKey和join都复用这个分区 rdd.reduceByKey(lambda a, b: a b)提示partitionBy本身是宽依赖会触发一次Shuffle。但如果后续有多个宽依赖操作这一次Shuffle可以省掉后面多次Shuffle总体是划算的。7. 从依赖关系看Spark与MapReduce的设计差异理解了RDD的依赖关系你就能明白Spark为什么比MapReduce快。MapReduce只有Map和Reduce两个阶段中间必须落盘而且每个MapReduce作业都要重新读取数据。Spark通过窄依赖的Pipeline执行把多个操作合并到一个Stage里数据在内存中流转不需要每一步都落盘。更重要的是Spark的DAG调度器可以根据依赖关系做全局优化。比如它可以识别出哪些操作可以合并、哪些分区可以跳过、哪些数据可以缓存复用。MapReduce的固定两阶段模型做不到这些。当然Spark的宽依赖仍然需要Shuffle和落盘所以它并不是完全替代MapReduce而是在迭代计算和复杂DAG场景下优势明显。我在实际选型时的经验是如果是简单的ETLMapReduce和Spark差别不大如果是机器学习迭代、图计算、交互式查询Spark的依赖关系模型带来的性能优势非常显著。但前提是你得理解依赖关系知道怎么切Stage、怎么减少Shuffle、怎么处理倾斜否则Spark也可能跑得比MapReduce还慢。最后分享一个我常用的检查清单每次写完Spark作业后过一遍能避开大部分依赖关系相关的坑宽依赖操作是否可以用map端预聚合替代groupByKey → reduceByKey/combineByKey是否有多个宽依赖操作可以复用同一个分区器partitionBy宽依赖之前的数据量是否已经尽量过滤和聚合分区数是否合理集群核心数的2-3倍是否有数据倾斜观察Task Duration的Max/Median比值Lineage是否过长超过20步考虑Checkpoint是否有重复计算的RDD需要Cache这套清单帮我省下了大量调优时间希望对你有用。
