★ FEATURED ARTICLE 分区管理这东西说大不大说小不小但绝大多数 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 涨到 5000Task 总数涨了五倍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 shuffletrue 时的 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 文件。这种做法在小数据量的场景下没有毛病。但如果数据量有几十 GBcoalesce(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.sizePython 里用rdd.getNumPartitions()。如果你想看每个分区里到底有多少条数据可以这样rdd.glom().map(_.length).collect().foreach(println)glom 会把每个分区的数据收集成一个数组然后看看数组的长度分布。这个方法在做分区调优前和调优后对比非常实用能直观看到每个分区的数据量是否均匀。但要注意 collect 会把所有数据拉到 Driver 端数据量大的时候不要轻易用。可以用采样或者只打印几个分区的数量来做初步判断避免 Driver 端内存爆掉。5.4 一张速查表总结选型日常开发里我基本按照下面这张表来选型。它不一定覆盖所有场景但至少能解决掉八成以上的问题场景推荐操作原因过滤后数据量变小只需要降分区coalesce(n)无 shuffle开销小分区数太少需要提高并行度repartition(n) 或 coalesce(n, shuffle true)需要洗牌才能重新均匀打散数据倾斜需要重新打散repartition(n)哈希重分布缓解分区不均输出单个文件coalesce(1)数据量大时考虑 shuffletrue保证单文件但要注意内存风险后续要 reduceByKey / join想避免二次洗牌partitionBy(new HashPartitioner(n))预分区让相同 key 落在同一分区只是想少生成文件不想改变计算并行度降低最后写入时的分区数而不是提前缩减避免在计算链条里引入额外的倾斜和瓶颈这张表的核心原则是能不做洗牌就不做洗牌但洗牌是为了解决倾斜和并行度不足的问题该洗牌的时候不要吝啬。两个目标——省资源、保均衡——在大多数时候是冲突的你要根据任务的实际瓶颈去权衡到底优先哪个。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再问自己数据分布均不均匀最后再决定写哪一个。
阅读完成 · 觉得有帮助？
