接手过 Spark 任务的人，基本都绕不开coalesce和repartition这两个算子。表面上看，一个是缩减分区，一个是扩大分区，好像“一减一增”就完事了。但实际在调优过程中，这两个算子背后隐藏的 shuffle 机制、窄依赖与宽依赖的区别、数据分布均衡性，才是真正决定作业性能的关键。不少初学者在 filter 之后数据量明显变小，直接coalesce一下发现任务还是慢；也有人想扩大并行度，用了coalesce(1000)，结果分区数纹丝不动。这篇文章就把这两个算子的底层逻辑、适用场景、参数估算方法和实战排坑过程讲透，给正准备优化 Spark 作业的朋友一个可以直接照着用的参考。
先给自己定个位：这篇内容适合两类人。一类是刚写完 Spark 作业、能跑但性能不理想，想弄明白怎么调整并行度的开发者；另一类是已经踩过coalesce或repartition的坑，想搞清楚“为什么这么写没用/更慢了”的进阶学习者。
1. 为什么分区数量会成为性能瓶颈
1.1 分区数决定了并行度与资源利用率
RDD 的分区数是 Spark 并行计算的最小单位。每个分区最终会对应一个 task，而 task 是真正被 executor 上的核心执行的任务单元。假设你申请了 10 个 executor，每个 executor 分配 2 个核心，那么同时能跑的 task 上限就是 20。如果 RDD 只有 10 个分区，无论集群资源多充裕，并行度最高也只有 10，剩下的资源全部闲置。反过来，如果 RDD 有 200 个分区，但集群总核心数只有 20，任务就会被切成 200 个小块分 10 批执行，调度开销上升，但某些场景下也未必坏——关键看单分区数据量和总数据量。
这里需要形成一个基本认知：分区数不是越大越好，也不是越小越好，而是要和集群资源、单分区数据量匹配。理想情况是一个分区处理的数据量大致在 128MB 到 256MB 之间（类似 HDFS 块大小的量级），这样每个 task 的启动、序列化、结果收集开销占比合理，又不至于因为分区太碎导致调度压力。
我在实际项目里见到过不少“分区病”案例：某个 RDD 数据量只有 500MB，却保留了 2000 个分区，每个 task 只处理 256KB 数据，光 task 启动和调度就占了作业时间的大半；反过来，某个 join 操作后单个分区数据量超过几个 GB，单个 task 直接 OOM。这两种情况都说明分区数没有跟着数据形态动态调整。
1.2 数据倾斜、小文件、空任务这些“分区病”怎么来的
分区数量不合理只是一个表象，更底层的原因是数据在分区之间分布不均衡。常见触发场景有三个：
第一，filter 操作之后。假设原始 RDD 有 1000 个分区，数据均匀分布，但 filter 一下筛掉了 90% 的数据，剩下 10% 的数据还留在 1000 个分区里。其中一部分分区可能完全没有任何数据，变成空 task；另一部分分区可能恰好保留了较多数据，导致后续处理时个别 task 数据量偏大。
第二，key 分布本身不均匀。如果业务数据的 key 有明显的热点，比如某个用户 ID 的数据量是其他用户的几十倍，那么即使分区数合理，经过 hash 分布后热点 key 所在分区的数据量也会严重偏大，这就是典型的数据倾斜。
第三，写入文件系统时的小文件问题。许多业务场景需要把计算结果写回 HDFS 或对象存储。当 RDD 分区数为 5000 时，直接写文件就会产生 5000 个小文件，不仅浪费存储，下游读取这些文件时也会产生大量额外开销。
这些问题的共同解决思路是调整分区数或重新分布数据。于是coalesce和repartition就成了最直接的工具。但在使用之前，必须搞清楚这两个算子内部到底做了什么。
2. coalesce 与 repartition 的底层逻辑差异
2.1 coalesce 的窄依赖：省钱但可能不均
coalesce方法的签名是coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)。注意默认参数shuffle = false，也就是说默认不走 shuffle，只是把父 RDD 的多个分区在逻辑上合并到一起。
从依赖关系来看，coalesce默认是窄依赖：子 RDD 的每个分区只依赖父 RDD 的一部分分区，且这个依赖过程不需要跨节点传输数据。它的实现方式是把多个父分区分配给同一个 task 处理，这样一来 task 数量减少了，但是每个 task 需要处理的数据量相应增大。
举个例子，假设一个 RDD 有 100 个分区，你想合并成 10 个分区。Spark 会尽量把每 10 个相邻的分区分配给同一个 task。如果这 100 个分区数据量均匀，结果还算理想；但如果某些父分区数据量已经是其他分区的几倍，合并后对应的新分区自然会变大，最终表现为“合并完反而更倾斜了”。极端情况下，比如从 2000 个分区合并到 10 个分区，个别 task 要读取几百个父分区的数据，而另一个 task 只读取几个分区，慢的任务会拖垮整个 stage。
coalesce默认模式的最大优势是不产生 shuffle 写磁盘和网络传输，性能开销极小。但它的前提是父分区数据分布足够均匀，或者合并比例不太夸张。一旦合并比例过大，或者数据本身分布不均匀，默认的coalesce就很容易踩坑。
2.2 repartition 的宽依赖：花钱买均匀
repartition的签名是repartition(numPartitions: Int): RDD[T]。从源码实现来看，它并不是一个独立实现，而是直接调用coalesce(numPartitions, shuffle = true)。也就是说，repartition等价于强制开启 shuffle 的coalesce。
开启 shuffle 之后，整个流程就完全不一样了。数据不再简单合并，而是会被重新打散，再按照分区算法（默认是 hash partitioner）分配到新的分区中。这意味着每个父分区里的数据都可能被拆开，通过网络传输到不同的 executor 节点，最终形成的新分区在数据量上基本均匀。
用生活化的方式理解：coalesce默认像是把几个小房间用墙打通合成一个大房间，房间总面积不变但房间大小取决于原始布局；repartition则是把所有房间里的东西全部倒出来，按照新的房间数量重新平均分配。后者一定能保证房间大小均匀，但代价是搬运东西的整个过程非常耗时费力。
由于 shuffle 过程涉及磁盘读写、序列化、网络传输，repartition的代价明显比coalesce高。所以一个常见原则是：能减少分区且父分区数据相对均匀时，优先用coalesce；需要增加分区、或者需要对分区数据做重新均衡时，用repartition。
2.3 依赖类型与 DAG 表现：repartition 就是 coalesce 的 shuffle 版本
如果从 DAG 依赖的角度观察，coalesce默认模式不会引入新的 stage，因为它不产生 shuffle 依赖，DAG 里只是多了一个窄依赖的 RDD 节点；而repartition会产生 ShuffleDependency，DAG 里会划分出新的 stage，在 Spark UI 上能看到明显的Shuffle Write和Shuffle Read记录。
判断一个算子是否引入 stage，最直接的方法是打印依赖关系：
val rdd = sc.textFile("hdfs://path/to/data", 100) val coalesced = rdd.coalesce(10) val repartitioned = rdd.repartition(200) println(coalesced.dependencies) // 输出中会出现窄依赖，例如 OneToOneDependency 或 PruneDependency 的变体 println(repartitioned.dependencies) // 输出中会出现 ShuffleDependency，这就是 stage 划分的标志这一节想强调的核心是：repartition不是coalesce的“加强版”，而是“shuffle 版”。两者在部分场景下可以互相替代，但代价和效果不同。搞清楚这一点，后面选型才不容易出错。
3. 实操：什么时候用 coalesce，什么时候用 repartition
3.1 缩减分区的正确姿势
缩减分区通常发生在三类场景：filter 之后数据量明显变小、计算结束准备写文件、缓存需要复用但不想保留过多分区。这时优先考虑coalesce，因为它不做 shuffle，代价极小。
但直接coalesce(n)并不总是安全。一个更稳妥的做法是：先评估数据分布是否均匀。如果数据本身比较均匀，比如日志类数据按时间分桶后过滤，父分区间数据量差异不大，那么直接coalesce没问题。如果经过某些算子后数据分布已经明显不均匀，或者合并比例超过 10 倍，建议使用coalesce(n, shuffle = true)，这等同于repartition(n)，虽然多一次 shuffle，但能保证合并后的分区大致均匀。
下面是一个典型的 filter 后缩减分区的例子：
val raw = sc.textFile("hdfs://nameservice/data/2024/06/*.log", 500) // 过滤掉大部分无效日志 val valid = raw.filter(line => line.contains("level=ERROR")) // 剩余数据可能只有原来的 5%，分布可能不均匀 val result = valid.coalesce(50, shuffle = true) // 因为经历了 filter，父分区数据量差异较大，开启 shuffle 更均衡这段代码里，为什么不是valid.coalesce(50)？因为 filter 之后很难保证每个父分区剩余的数据量一致，如果某个分区的错误日志明显多于其他分区，默认合并后仍然倾斜。shuffle = true之后，数据重新打散，Stage 再往下执行时每个分区处理量更接近。
还有一种常见需求是写文件前缩减分区，减轻小文件问题：
val processed = ... // 经过一系列计算，分区数可能有 2000 processed.coalesce(100).write.mode("overwrite").parquet("hdfs://output/path")这里如果分区之间的数据量本来较均匀，用默认 coalesce 即可。需要注意的是，写入文件的分区数直接决定了输出文件数量，所以目标分区数应该结合下游读取的并发度来决定，而不是越少越好。
3.2 扩大分区的唯一正路
扩大分区只有一个选项：repartition。因为coalesce默认模式在目标分区数大于当前分区数时，效果并不靠谱。你调用rdd.coalesce(2000)时，如果当前 RDD 只有 200 个分区，实际结果并不会得到 2000 个分区。原因是 coalesce 的默认实现需要把多个父分区合并到同一个子分区，父分区数量少于目标分区数时，它无法凭空拆分出更多分区。
扩大分区的场景主要有两个：一是后续计算依赖更大并行度，比如集群资源充足但 RDD 分区数太少，导致并行度不足；二是需要消除窄依赖带来的数据倾斜，通过repartition重新打散数据。
// 分区数过少，并行度不足 val underPartitioned = sc.textFile("hdfs://path/to/medium", 20) val rdd = underPartitioned.repartition(200) // 200个分区对应集群总核心数200，task并发能力提升这里要注意，扩大分区之前先想清楚一个问题：扩大分区带来的并行度提升是否能抵消 shuffle 的开销。单个 RDD 如果有 20 个分区，但集群有 200 个核心，直接 repartition 到 200 确实能让后续计算并行度提高，但中间这段 shuffle 本身要读取 20 个分区的数据、写入 200 个分区，消耗的时间和资源也需要计入总成本。如果只是临时计算一次，不一定划算；如果这个 RDD 会被多个后续计算复用，那这次 shuffle 就是值得的。
3.3 参数怎么定：按数据量、按核心数、按目标大小
分区数量的设定没有万能公式，但有几种经过实践检验的估算思路。
第一种，按单分区目标数据量估算。假设总数据量为 10GB，希望每个分区处理 128MB 到 256MB，那么目标分区数大约在 40 到 80 之间。这个办法适合数据量明确、希望控制单 task 内存压力的场景。
// 估算举例：总数据量约10GB，单分区目标128MB val targetPartitions = 10 * 1024 / 128 // 约80第二种，按集群并行度估算。假设集群有 10 个 executor，每个 4 核，总核心数 40。那么分区数设为核心数的 2 到 3 倍，也就是 80 到 120，可以在一定程度上减少 task 执行不均匀带来的尾部延迟。这种办法适合不知道精确数据量、但清楚资源配额的场景。
第三种，按下游消费需求估算。如果写文件，分区数应该参考下游读取并发度；如果后续 join，分区数要考虑参与 join 的两侧 RDD 分区数，或者直接利用 join 算子的分区参数来设置。这里有个技巧，就是尽量把分区调整和 shuffle 算子合并，减少额外的 shuffle 次数。比如reduceByKey(numPartitions)本质上就是在 shuffle 阶段完成分区调整，这比执行reduceByKey()之后再做一次repartition更高效。
// 不推荐：先shuffle再额外repartition val rdd = words.map((_, 1)).reduceByKey(_ + _).repartition(100) // 推荐：直接在shuffle阶段指定分区数 val rdd = words.map((_, 1)).reduceByKey(_ + _, 100)第二种写法多了一步 reducer 归并之后的分区调整，而第一种写法在 reduceByKey 的 shuffle 阶段就直接输出到 100 个分区。虽然结果一样，但实现的执行计划更精简，少了一次完整的 shuffle。
4. 踩坑记录与排查技巧
4.1 coalesce 后数据倾斜的经典现场
一次处理用户行为日志的离线任务中，原始数据按天分桶，共有 800 个分区。经过一个复杂的过滤逻辑后，剩余数据量只有原来的 3%，于是直接用coalesce(20)缩减分区。结果跑起来后，20 个 task 中有一个执行了 40 分钟，而其他 task 基本在 2 分钟内结束，整个 stage 被严重拖慢。
排查时先看 Spark UI 的 Stage 详情，发现那个高耗时 task 的Shuffle Read和输入记录数是其他 task 的 30 倍。原因就是 filter 后某些父分区保留了更多数据，coalesce 合并时没有打散重分布。解决办法很简单，改成coalesce(20, shuffle = true)，让数据重新均匀分布，整个 stage 最后 5 分钟跑完。
这个案例说明了一个容易被忽略的道理：coalesce 默认的“零开销”是有前提的。如果节省了一次 shuffle 却制造了一个数据倾斜的拖后腿 task，整体收益可能是负的。实际开发中我通常会把 filter 之后的分区缩减直接写成coalesce(n, shuffle = true)，省得反复试错。
4.2 repartition 之后反而变慢了
另一个相反的案例：某任务为了加大并行度，把 100 个分区的 RDDrepartition(1000)，结果作业整体耗时反而从原来的 15 分钟涨到了 25 分钟。看 Spark UI 后发现，Shuffle Write 和 Shuffle Read 的数据量巨大，shuffle 阶段的耗时占了 60% 以上，而真正受益于高并发的后续计算只有很短的执行时间。
这个问题的关键不是 repartition 本身错了，而是收益和成本不成比例。如果一个 RDD 后续只被连续的窄依赖算子使用，一次完成的计算总耗时本来就短，那么额外引入 shuffle 就是把大量时间花在重新分配数据上，得不偿失。正确的做法是评估 RDD 被后续计算的使用频率和数据依赖深度。如果 RDD 会被多个后续算子多次使用，比如要 join、要 group by、还要写多个输出，那么一次 repartition 的收益会被摊薄到多项计算里，此时扩大分区才划算。
4.3 coalesce 扩大分区的无效操作
有一个高频错误是把coalesce当成“既能减又能增”的万能算子，直接写rdd.coalesce(2000)想扩大分区。结果通过rdd.getNumPartitions查看，分区数还是原来的，或者根本没有达到预期。原因是默认 shuffle=false 时，coalesce 无法在父分区数量不足的情况下凭空拆分出更多分区，这个行为在 Spark 的 RDD 继承关系中是受限的。
如果在生产代码里发现这种情况，直接用repartition(2000)替换就行。但如果想过把瘾研究源码，可以看一眼CoalescedRDD的实现，它会基于父 RDD 的分区数组做合并映射，请求的分区数大于父分区数时，映射关系无法产生更多有效分区，自然就“无效果”。这个知识点知道即可，实际开发不用纠结实现细节，记得一个结论就好了：要扩分区，用 repartition。
4.4 用 Spark UI 和依赖链快速定位分区问题
排查分区相关问题，我总是先看两个地方。第一是 Spark UI 的 Stage 列表，重点观察每个 stage 的 task 数量、Shuffle Write/Read 大小，以及有没有明显超出平均耗时的 task。如果存在单个 task 的执行时间远大于中位数，多半是分区数据不均衡。第二是rdd.toDebugString，打印出来的缩进依赖树能直观区分窄依赖和宽依赖，看清某个操作前后是否引入了新的 stage。
举一个排查的完整例子：
rdd.toDebugString // 输出示例 // (2) MapPartitionsRDD[30] at coalesce at ... // | ShuffledRDD[29] at repartition at ... // +-(2) MapPartitionsRDD[28] at map ...在 debug 字符串中看到ShuffledRDD就代表这里发生过 shuffle。如果某些 RDD 阶段前多了 ShuffledRDD 而你没有主动调用过 repartition，那可能是 groupByKey、join、distinct 等算子内部产生的。熟悉这个过程后，你就能快速判断哪些 shuffle 是业务必要、哪些是操作失误造成的额外开销。
另外一个常见的排查技巧是在开发阶段用getNumPartitions和glom().map(_.size)来检查每个分区的记录数分布：
val partitionSizes = rdd.glom().map(_.length).collect() println(partitionSizes.mkString(", ")) // 如果数值差异巨大，例如 [1, 10000, 2, 80000, 3]，说明分区数据严重不均衡这个检查方法成本很低，适合在调优前先摸清分区分布状况，再用 coalesce 或 repartition 对症下药。
5. 几个经验性的建议
一路写到这里，想结合个人实践经验给几条务实建议。
第一，coalesce和repartition不是简单的“少用贵、多用便宜”的关系。在需要缩减分区时，默认 coalesce 的零 shuffle 优势很明显；但如果父分区数据分布不均匀，宁可使用 shuffle=true，也不要为了省 shuffle 而制造数据倾斜。
第二，分区调整尽量和 shuffle 算子合并。reduceByKey(100)比reduceByKey().coalesce(100)省一次全量 shuffle，join时指定分区的逻辑也类似。每减少一次 shuffle，作业的磁盘和网络开销都会显著下降。
第三，调优时要看整体执行计划，不要只盯着一个算子。有时分区数合理了，但上游数据源读入时分区数设计不合理，照样会影响作业。通常我会从数据读入的第一层就开始估算分区数量，中间再根据数据量的变化动态调整一到两次。
第四，对于线上任务，改完分区策略后一定要对照 Spark UI 观察前后两版运行的 Stage 耗时。真实分布式环境中的网络抖动和资源竞争很难在本地复现，只有实际跑一遍才能确认改动有效。
最后再说一个小技巧。如果你不确定当前 RDD 到底该用 coalesce 还是 repartition，可以在本地集群上用小数据集分别跑一遍，打印toDebugString和每个分区的记录数，观察 shuffle 前后分区分布的变化。这个验证过程花不了几分钟，但能让你对这两个算子的行为形成非常直观的印象。我在带新人的时候，就经常让他们自己去跑一遍这个对比实验，效果比解释十遍都强。
