企业官网建设流程全解析

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。

我画个简单的对比表帮你记忆：

2.3 一个容易搞混的案例：join到底是宽依赖还是窄依赖

很多人以为join一定是宽依赖，其实不一定。如果两个RDD在join之前已经按照相同的分区器（Partitioner）进行了分区，并且分区数量一致，那么Spark可以做到co-partitioned join，这时候每个子RDD分区只需要从对应的父RDD分区拉取数据，不需要全量Shuffle，这就是窄依赖。

但如果你没有提前做好分区，直接对两个普通RDD做join，Spark默认会走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如何根据依赖关系切分Stage

3.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归为一个Stage（Stage 0），rdd3、rdd4归为另一个Stage（Stage 1）。Stage 0负责读取数据并做filter和map，然后Shuffle输出；Stage 1负责接收Shuffle数据做reduceByKey和map，最后collect。

3.2 Stage类型：ShuffleMapStage与ResultStage

Spark里有两种Stage：ShuffleMapStage和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是宽依赖，会触发Shuffle；coalesce如果不开启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 Lineage：Spark的“后悔药”机制

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端做预聚合（combine），Shuffle数据量直接降了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(f"Partition {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之后直接做reduceByKey，Shuffle时会有大量空分区参与，浪费资源。我通常会在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场景下优势明显。

我在实际选型时的经验是：如果是简单的ETL，MapReduce和Spark差别不大；如果是机器学习迭代、图计算、交互式查询，Spark的依赖关系模型带来的性能优势非常显著。但前提是你得理解依赖关系，知道怎么切Stage、怎么减少Shuffle、怎么处理倾斜，否则Spark也可能跑得比MapReduce还慢。

最后分享一个我常用的检查清单，每次写完Spark作业后过一遍，能避开大部分依赖关系相关的坑：

• 宽依赖操作是否可以用map端预聚合替代（groupByKey → reduceByKey/combineByKey）
• 是否有多个宽依赖操作可以复用同一个分区器（partitionBy）
• 宽依赖之前的数据量是否已经尽量过滤和聚合
• 分区数是否合理（集群核心数的2-3倍）
• 是否有数据倾斜（观察Task Duration的Max/Median比值）
• Lineage是否过长（超过20步考虑Checkpoint）
• 是否有重复计算的RDD需要Cache

这套清单帮我省下了大量调优时间，希望对你有用。