Spark分区管理：coalesce与repartition原理、选型与避坑指南

分区管理这东西，说大不大，说小不小，但绝大多数 Spark 作业卡死、内存爆掉、产出小文件成灾的时候，十有八九都和分区数脱不了干系。RDD 里调整分区最常用的两个算子就是 coalesce 和 repartition，名字长得像，底层行为却差了一整个 shuffle 的距离。这篇文章就把这两个算子彻底掰开揉碎，从原理到实操，从选型到避坑，一次讲透。

我最早接触这两个算子的时候，挺天真地以为 coalesce 就是缩小分区、repartition 就是扩大分区，各干各的活。直到有一次，一个跑了三小时的离线任务，我为了收拢数据用了 coalesce(10)，结果某个 Executor 直接 OOM，排查了大半天才发现是过度缩减分区引发的数据倾斜。从那以后我明白了一个道理：分区操作不是改个数字那么简单，它背后牵涉的是 shuffle、磁盘 IO、内存模型和任务调度一整条链路。

1. 为什么分区数量和 Spark 任务表现强相关

1.1 分区数如何决定任务的并发上限

Spark 最核心的抽象是 RDD，而每个 RDD 内部都维护着一个分区列表。分区是数据的最小逻辑切片，也是 Spark 调度任务的最小粒度。这么说吧，一个 Stage 内的并行度，原则上就等于该 Stage 最后一个 RDD 的分区数。每个分区会被打包成一个 Task 交给 Executor 上的某个线程去跑。

所以分区数大了，Task 数量就多，集群资源被利用得更充分；分区数小了，哪怕你有 100 个 Executor，也只会少数几个在干活，其余全部空闲。这就像开了一家有十个窗口的银行，但营业员只开放两个窗口，队伍排得再长，也就那两个窗口在处理业务，吞吐量上不去。

我在实际生产里见过一个很典型的案例：某个报表任务，输入数据只有 200 个分区，但集群配置了 40 个 Executor、每 Executor 4 核。理论上并行度可以到 160，结果因为分区数 200，数据分布不均匀，部分 Task 很快就跑完了，少部分 Task 拖了整个 Stage 的后腿，最终运行时间比调优后多了将近一倍。这就是分区数和集群算力不匹配的代价。

1.2 分区过少和过多分别是什么体验

分区过少的问题很直观：并行度不足、CPU 闲置、单个 Task 处理的数据量过大。尤其在数据倾斜的情况下，某一个分区里堆积了大半数据，那个 Task 就要承受远超平均水平的计算压力，轻则拖慢整体进度，重则直接内存溢出。

分区过多也不是什么好事。每个 Task 都有固定的调度开销、序列化开销和结果物化开销。分区数从 1000 涨到 5000，Task 总数涨了五倍，Executor 的线程调度、Driver 的 Task 调度队列都会承压。更麻烦的是，如果每个分区里的数据量都很小，比如只有几十 KB，那大量时间都花在了调度和启动上，真正算数据的时间占比反而下降。

此外，分区数还会直接影响最终输出文件的数量。没有特殊处理的话，每个分区会写一个文件，分区数等于多少，小文件就有多少。下游如果是 Hive 表或者需要被其他引擎读取，小文件成灾会让元数据服务和查询性能双双恶化。控制分区数，本质上就是在计算并行度和资源开销、文件粒度之间找一个平衡点。

2. 搞懂 coalesce 和 repartition 的本质区别

2.1 coalesce：默认不洗牌，只做分区合并

coalesce 的默认实现是 shuffle 参数等于 false，也就是不触发 shuffle，纯做分区合并。它的思路是尽量把多个分区的数据留在原Executor节点上，通过重新映射的方式把相邻的几个逻辑分区并成一个。

举个例子，当前 RDD 有 100 个分区，执行 coalesce(50) 之后，Spark 会把两个相邻的分区合并为一个。关键是这个合并过程不涉及跨节点数据搬移，数据还在原来的 Executor 上，只是逻辑上归属的分区变了。

这种设计的直接好处是开销极小。没有 shuffle，就没有磁盘写、网络传输和磁盘读这三部曲，合并过程非常轻量。但是代价也很明显：如果原来的分区数据量就不均匀，比如第一个分区 10GB、第二个分区只有 100MB，把这两个分区合并后，新的分区仍然是严重倾斜的。coalesce 只是把分区数变小，不做数据重分布。

还有一个容易被忽略的点：coalesce 合并后，每个节点上可能会残留空分区。假设 4 个 Executor 各持有 2 个分区，全部 coalesce 到 1 个分区，那只会保留其中一个分区的数据，其他分区变成空壳。空分区不会触发真正的计算，看起来没什么影响，但如果下游逻辑依赖分区内数据做累加或聚合，空分区会导致结果缺失。

2.2 repartition：一定会洗牌，重新打散数据

repartition 的全名其实可以理解为 coalesce(numPartitions, shuffle = true)，它的本质就是触发一次完整的 shuffle，把全部数据按照新的分区数重新哈希打散。

洗牌的过程大致是这样的：每个上游 Task 根据自己的数据，通过一个分区器计算每条记录应该归属到哪个新的分区，然后把数据写到本地磁盘的临时文件里。下游的 Task 从所有上游节点的临时文件中把属于自己那一份数据拉回来，再做归并和聚合。整个过程涉及磁盘写、网络传输和数据归并，开销比 coalesce 大得多。

但洗牌带来的好处也非常明确，那就是数据分布会被重新打乱，原来倾斜的某个分区会被打散到多个新分区里，每个新分区的数据量理论上更均匀。repartition 默认使用的分区器是哈希分区器，根据 key 的哈希值取模，所以相同 key 的数据还是会进到同一个分区。如果你的目的是让相同 key 的数据在后续 join 或 groupBy 时落到同一个分区，这种特性反而是你想要的。

我个人的理解是，repartition 更像是一次数据重组，代价大，但能换来数据分布的重构；coalesce 更像是一次数据归拢，代价小，但解决不了分布不均的问题。把这个底层逻辑刻在脑子里，选型的时候就不容易犯迷糊。

2.3 shuffle=true 时的 coalesce 和 repartition 是同一件事吗

这个问题很多初学者问过我。从源码层面看，repartition 的实现就是调用coalesce(numPartitions, shuffle = true)，两者在行为上确实等价。

但注意两个细节。第一，代码可读性层面，repartition 语义更清晰，看到 repartition 就知道这里发生了 shuffle，看到 coalesce 第一反应是无 shuffle，如果加了 shuffle = true 反而容易让读代码的人产生疑惑。所以我的习惯是：需要洗牌重分布一律写 repartition，纯合并分区一律写 coalesce。

第二，coalesce 有一个特殊能力是 repartition 不具备的，那就是它可以接受目标分区数大于当前分区数，但前提是 shuffle 参数必须为 true。如果写成coalesce(200, shuffle = false)而当前只有 50 个分区，它不会报错，但也不会把分区数变成 200，实际结果还是 50。源码里对这种情况有处理，无 shuffle 路径下目标分区数大于当前分区数时会直接返回原 RDD，或者最多按当前分区数处理。

接着聊一个更细的差异。coalesce 在无 shuffle 时，实际生成的分区数并不总是精确等于你要的数字。比如你从 100 个分区 coalesce 到 30，而集群有 3 个节点，每个节点上恰好 33 个分区，那么合并后很可能每个节点产出 1 个分区，总共 3 个分区，而不是 30 个。因为无 shuffle 的 coalesce 是按节点粒度做合并的，同一个节点上的多个分区分到同一个组里，最终分区数大约等于参与计算的节点数。这一点在生产环境里特别容易让人困惑，后面我会在常见问题里再展开。

3. 实操：coalesce 缩减分区的正确姿势

3.1 最简单的缩减：过滤后收拢数据

coalesce 最常见的应用场景是过滤操作之后。比如你从 Hive 里读了一张大表，然后做了一次条件过滤，过滤掉的数据占比可能高达百分之九十。这时候 RDD 的分区数还是读取时候的值，比如 1000，但每个分区里剩下的数据只有一点点。1000 个空荡荡的分区带来的是 1000 个 Task 的调度开销，纯属浪费。

正确的做法是在过滤之后紧跟着一个coalesce(100)，把分区数降到和剩余数据量匹配的量级。因为过滤没有产生 shuffle，数据分布仍然是原来每个分区各自持有的数据，只是量变少了，所以用无 shuffle 的 coalesce 完全没有问题，还能省掉一次洗牌的开销。

实际代码大概是这样的：

val rawRdd = sc.textFile("hdfs:///data/raw_logs") val filteredRdd = rawRdd .filter(line => line.contains("click") && line.split(",").length > 8) .map(parseLine) .coalesce(100)

这里 coalesce(100) 安排在 map 之后，此时数据已经完成了解析，分区收拢不会影响计算逻辑。收拢之后，后续的存储、计算或者进一步的算子操作，都基于这 100 个分区展开，调度压力小了很多。

要注意的一点是，过滤后收拢分区的目标数量不能拍脑袋。我的经验是估算一下剩余数据总量，比如过滤后还剩 20GB，每个分区期望处理 200MB 左右，那就设置 100 个分区左右。这个比例不是固定的，和每条记录的复杂度、是否有聚合操作都有关，但 100 到 200MB 每分区是一个比较安全的起点。

3.2 缩减到 1 个分区：写小文件的常见做法

coalesce(1) 可能是最常被写进代码的分区操作之一。它常常出现在需要把计算结果汇总成一个文件再交给下游的场景里，比如生成一个全量的导出文件，或者写入某个对文件数量有要求的目录。

rdd.coalesce(1).saveAsTextFile("hdfs:///tmp/output")可以保证输出目录里只有一个 part 文件。这种做法在小数据量的场景下没有毛病。但如果数据量有几十 GB，coalesce(1) 就非常危险。

为什么危险？因为无 shuffle 的 coalesce(1) 虽然不做数据搬移，但最终所有数据都会归属到那一个分区对应的 Task 上去执行输出。这个 Task 要串行地处理全部几十 GB 的数据，既考验它的计算能力，也考验它的内存承受力，而且那个 Executor 上还同时承担了其他 Task 的压力。我在生产里见过很多次，任务最后一步 coalesce(1) 把整个作业拖死，或者导致 Executor 直接崩溃。

如果数据量大又必须输出单个文件，我的建议是退一步，考虑用 coalesce(1, shuffle = true)，虽然这样做会把全部数据洗牌到一个分区，在磁盘和网络上有一轮完整的数据搬移，但至少数据是均匀集中归拢的，不会因为原本某个分区数据量过大而直接撞上内存红线。再或者，也可以考虑输出多个文件再在文件系统层面合并，比如用 HDFS 的 getmerge 命令把多个 part 文件合并成一个。

3.3 必须避免的坑：激进缩减带来的数据倾斜

这一节是我最想强调的。coalesce 无 shuffle 的合并逻辑是相邻分区直接拼在一起，它完全不考虑分区之间的数据量是否均衡。假设你有 10 个分区，数据量分布如下：第一个分区 8GB，其余 9 个分区每个 300MB。coalesce(5) 之后，第一个分区和第二个分区合并成 8.3GB 的新分区，而其他几个新分区都在 600MB 左右。倾斜依旧存在，甚至可能更严重。

更极端的情况是再往下缩减，比如从 10 个分区 coalesce(2)，很可能两个新分区里一个占了 8.3GB，另一个只有 1.7GB。这种倾斜在后期计算中会被无限放大，尤其是再叠加一次 reduceByKey 或者 join，倾斜分区上的 Task 会成为整个作业的瓶颈。

所以在做大幅缩减分区之前，先看数据分布。用rdd.glom().map(_.length)这样的方式观察每个分区的大小，如果发现数据本身就不均匀，就别再用无 shuffle 的 coalesce 了，改用 repartition 或者 coalesce(shuffle = true)。虽然多花一次 shuffle 的成本，但换来的是后续计算负载的均衡，从整体时间上看往往是划算的。

还有一个我在项目里踩过的坑：coalesce 之后紧接着的算子会导致 shuffle 的，分区数的设置容易被 shuffle 行为覆盖。比如rdd.coalesce(10).reduceByKey(_ + _)，reduceByKey 的默认分区数是根据父 RDD 的最大分区数来定的，而 coalesce(10) 之后父 RDD 分区数已经变成 10，所以 reduceByKey 的输出分区数就是 10，这里没问题。但如果你后续还有 repartition 或 join 等操作，最终分区数的决定权就不在 coalesce 手里了，需要在链条里再检查一次。

4. 实操：repartition 扩大分区的适用场景

4.1 数据倾斜后的重新打散

repartition 最典型的用途就是缓解数据倾斜。当某个 RDD 因为 key 分布不均，导致部分分区数据量远超其他分区时，一个直接的思路就是repartition(n)把数据重新哈希打散到更多的分区里。

举个例子，原始数据里有 100 万个 key，其中某个大 key 占了数据量的 60%。如果直接 groupByKey 或者 reduceByKey，那个大 key 所在的分区必然成为热点。这种情况下，先做rdd.repartition(200)并不能解决大 key 本身的问题——相同 key 经过哈希分区还是进同一个分区——但如果你只是需要整体均摊压力，比如先把数据做大范围的随机打散再计算，repartition 就有效果。

真正想处理大 key 倾斜，单纯靠 repartition 是不够的，还需要配合两阶段聚合或者加盐等手段。但 repartition 在另一个场景下非常有用：原始分区数太少，需要提高并行度。比如从 HDFS 读了一堆小文件，总共只有 20 个分区，而集群有 100 核在等着干活，这时候repartition(100)能让利用率直接拉满。

4.2 reduceByKey / join 之前的预分区

这里要区分一下 repartition 和 partitionBy 的差异。repartition 对 RDD 的数据本身做哈希重分布，分区数可以任意指定，但分区键默认用的还是原有 key。如果你想在 reduceByKey 之前让相同 key 的数据预先落到同一个分区，同时又把分区数调整到理想值，可以这样做：

val repartitioned = sourceRdd .map(kv => (extractKey(kv), kv)) .partitionBy(new HashPartitioner(200))

partitionBy 也是触发 shuffle 的，并且它会强制使用指定的分区器，后续的 reduceByKey 或 join 如果分区器一致，就能避免二次 shuffle，这个优化点在大型 join 作业里效果非常明显。

回到 repartition 本身，如果你只是在做rdd.flatMap(...).repartition(200)，那它的作用只是把数据分散到更多分区，分区键仍然是原 key 的哈希值。如果你的后续逻辑不依赖 key 的分区一致性，这么做完全够用。

4.3 从窄依赖跨到宽依赖时的选择

分区操作会改变 RDD 的血缘关系，影响后续 Stage 的划分。coalesce 无 shuffle 时，父子 RDD 的分区保持窄依赖关系，后续算子仍然可以和前面的操作合并在同一个 Stage 里，这样减少了 Stage 的数量，也就减少了调度开销。而 repartition 因为引入了 shuffle，会切断血缘上的窄依赖，强制产生一个宽依赖边界，后续的算子会进入新的 Stage。

这个特性有时候被用来做主动的 Stage 隔离。比如在一个复杂的血缘链条里，前一段逻辑非常耗时且数据膨胀严重，你希望在它结束之后做一次 checkpoint 或者缓存，避免后续失败时从头重算，那可以在合适位置插入一个rdd.persist()加rdp.repartition()的操作，让重算代价可控。

不过从另一个角度看，不要滥用 repartition。每多一次洗牌，全量数据就会在磁盘和网络上多走一轮。我见过有人为了凑分区数，在一条血缘里连续写了三个 repartition，每次都是 200 分区，对任务没有实质好处，却白白多了两轮全量洗牌。这种操作对性能的伤害是不可忽视的，分区和洗牌的次数一定要尽可能少。

5. 常见问题与排查技巧实录

5.1 为什么 coalesce(1000) 没有生效

一个很常见的疑问是：我明明写了 coalesce(1000)，为什么下游看到的分区数还是 200？

原因可能有几个。第一，当前 RDD 的分区数本来就小于 1000，无 shuffle 的 coalesce 不会增加分区数，所以实际分区数只能维持原有的值或者更少。第二，coalesce 无 shuffle 时的分区数是按节点粒度合并的，如果 Executor 数量少于目标分区数，可能得不出你期望的分区数。

我提供一个可靠的排查手法：在 coalesce 前后都打印rdd.partitions.size。如果发现前后分区数不对，先把 shuffle 参数显式设为 true 试试。比如coalesce(1000, shuffle = true)，这时候分区数一定精确变成 1000，因为有了完整的洗牌，数据被重新哈希分配到指定数量的分区中。

还有一种情况是在 coalesce 之后又接了一个会触发 shuffle 的高阶算子，比如 join 或 repartition，导致你查看的是下游 RDD 的分区数。这时候需要检查整条血缘链上每一个 RDD 的分区数，定位是哪个算子真正决定了最终分区数。

5.2 缩减后某个 Executor 内存炸了

这个问题的根源几乎都是数据倾斜。无 shuffle 的 coalesce 把多个分区的数据硬拼在一起，如果某个节点上的数据总量远超其他节点，那么这个节点对应的 Executor 在后续计算中就要承受超量数据。内存不够就溢出，磁盘来不及写就卡死。

排查思路是这样的：先看崩溃 Task 的日志，找到它所属的 Stage 和 Task ID，然后在 Spark UI 的 Stage 页面查看每个 Task 的 Shuffle Read 或 Input 大小。如果看到某个 Task 的输入量是其他 Task 的十几倍，那基本坐实了倾斜。

解法有两种。一种是把 coalesce 改成带着洗牌的方式，让数据重新均匀分布；另一种是干脆不要缩减分区，接受更多的分区数但保住数据均衡。很多时候，分区多的代价只是多了一些 Task 调度开销，但比单个 Executor 挂掉强得多。具体选哪种，得看你的数据量和节点数。

5.3 如何判断当前 RDD 的分区数

这是最基础的一个问题，但真的有人不知道。Scala 里用rdd.partitions.size，Python 里用rdd.getNumPartitions()。如果你想看每个分区里到底有多少条数据，可以这样：

rdd.glom().map(_.length).collect().foreach(println)

glom 会把每个分区的数据收集成一个数组，然后看看数组的长度分布。这个方法在做分区调优前和调优后对比非常实用，能直观看到每个分区的数据量是否均匀。

但要注意 collect 会把所有数据拉到 Driver 端，数据量大的时候不要轻易用。可以用采样或者只打印几个分区的数量来做初步判断，避免 Driver 端内存爆掉。

5.4 一张速查表总结选型

日常开发里，我基本按照下面这张表来选型。它不一定覆盖所有场景，但至少能解决掉八成以上的问题：

这张表的核心原则是：能不做洗牌就不做洗牌，但洗牌是为了解决倾斜和并行度不足的问题，该洗牌的时候不要吝啬。两个目标——省资源、保均衡——在大多数时候是冲突的，你要根据任务的实际瓶颈去权衡到底优先哪个。

6. 基于实操经验的几条心法

6.1 我的几条经验法则

第一，分区数的目标值不要拍脑袋。先算一下数据量，设定每个分区的目标处理量，比如 128MB 到 256MB，然后用总数据量除以目标量得到初始分区数。这不是一个精确公式，但比随手写个 10 或者 1000 靠谱得多。

第二，coalesce 用在过滤后和写出前是最安全的。过滤后数据量缩小，合并分区没有副作用；写出前合并分区可以减少文件数，只要数据量不大，coalesce(1) 完全没问题。中间的计算逻辑里除非明确知道数据分布是均匀的，否则不要动 coalesce 的念头。

第三，遇到性能问题先看 Spark UI。Stage 的时间分布、每个 Task 的数据量、shuffle 读写的大小，这些信息比任何猜想要有价值。分区相关问题在 UI 上会非常直观地表现出来：要么是某个 Task 独占了一个长尾，要么是 Task 数量多得离谱。

第四，永远不要连续对同一个 RDD 做多次无意义的重分区。每做一次分区调整，尤其是触发 shuffle 的那种，都会带来一轮全量数据的磁盘写入、网络传输和磁盘读取。一次优化最多加一个分区操作，再多就是损耗。

6.2 调优效果自测方法

调优完怎么验证自己改对了？我的建议是记录三个指标：总运行时间、每个 Task 的平均处理时间、以及最后生成的文件数量。调优前后对比这三个值，基本就能判断分区调整是否起到了正面效果。

还有一个细节值得留意：分区数调大之后，Task 数量变多，Driver 端的调度开销和 Executor 端的线程切换开销都会上升。如果调大分区后总时间反而变长，说明你的分区数已经超过了集群并行能力的上限。这时候不是继续加大分区数，而是回头检查 Executor 数量和每个 Executor 的核数，找到并发度的天花板。

分区调优这块，我踩过的坑不算少，上面的内容都是真金白银换来的教训。coalesce 和 repartition 看着简单，真正用好确实需要把原理吃透。下次再有人问你这两个算子怎么选，你可以直接告诉他：先问自己要不要 shuffle，再问自己数据分布均不均匀，最后再决定写哪一个。