Spark DataFrame重新分区实践：从数据倾斜到性能调优的完整指南

上个月排查一个跑批任务变慢的问题，最后定位到的根因让整个组都挺意外：数据量其实没怎么涨，但作业从十几分钟拖到了四十分钟。顺着Spark UI一个个stage看下去，发现某个宽依赖前的shuffle写阶段产生了大量的小task，每个task处理的数据量差异非常大，一部分task几秒就跑完了，另一部分task要拖几分钟。原因也很典型——前面的环节多次做DataFrame重新分区，但分区键选得随意，加上shuffle分区数一直用默认值，数据早就歪得不成样子。

这个经历让我决定把Spark DataFrame重新分区这件事完整地梳理一遍。分区是Spark作业性能里最容易被忽视、但影响面最大的一个环节，hzhz中间随便一个调整就可能让任务快一倍或者慢一倍。这篇文章不会只讲API怎么用，重点放在分区的底层机制、repartition和coalesce的关键差异、数据倾斜时怎么重新设计分区键、以及分区调整之后容易踩的隐藏雷区。

1. 一个慢任务的排查现场：分区混乱比数据量更可怕

先还原一下当时的场景。某个离线清洗链路，每天处理大约200G的日志数据，源表按天分区存储在数仓里，链路中间有几步join和filter。某一天开始任务耗时突然变长，最开始以为是数据量增长导致的，但核对了一下源头统计，数据量基本稳定。

我把Spark UI的stage列表调出来，发现一个很有意思的现象：某个stage的task数量是200，但不同task处理的数据量从几十MB到一个多GB都有，最慢的那个task硬生生拖了整个stage的后腿。这个现象叫数据倾斜，但根子在于分区本身就没分好。

再往前翻血统，发现这个链路里至少有两个环节对DataFrame做了重分区操作。第一处是读取源表后为了跟另一张事实表做join，直接把DataFrame repartition到了某个业务字段上。第二处是某位同事为了“让下游并行度更高一些”，随手写了一个repartition(500)。两处操作叠加之后，数据的物理分布已经严重偏离合理状态。

那个业务字段本身就有严重的偏态分布，部分枚举值的记录数占了总量的大头。用这样的字段做重分区键，相当于把所有数据按一把严重失衡的尺子切了一遍，某几个分区被塞得满满当当，其他分区几乎空着。并行度再高也没用，整个stage被那几个满载的分区卡死。

这次排查让我意识到一个容易被忽略的事实：重新分区的目的不是“让DataFrame分区更多或更少”，而是“让数据在集群上均匀、合理地分布，同时让下游计算的开销最小”。分区本身只是手段，数据分布质量和后续操作的适配度才是关键。分区键选错、分区数拍脑袋定、repartition和coalesce不分场合乱用，这三类问题几乎是性能劣化的共同源头。

2. 重新分区时Spark到底在做什么：shuffle、分区器与执行计划

想用好重新分区，得先搞明白Spark内部为了重排数据到底付出了什么代价。很多人把repartition当成一个轻量级操作，实际上它是一个会触发全量shuffle的宽依赖操作，代价不比一次大join小多少。

2.1 shuffle的本质：数据被打散、落盘、再拉取

shuffle是Spark里成本最高的操作之一。repartition的时候，每个分区里的数据要先按目标分区规则算好去往哪个新分区，然后写到本地磁盘上进行排序和聚合，等待下游stage的task来拉取。这个过程中涉及大量的磁盘I/O、网络传输和序列化开销。如果数据量是TB级别，一次repartition的shuffle数据量可能轻松超过原始数据量的总和，因为一份数据在map端和reduce端都完整过了一遍。

这也是为什么repartition不能随便用。每调用一次repartition，就相当于让全量数据在集群里“搬一次家”。搬家本身没问题，但频繁搬家、无序搬家，性能必然崩。

2.2 分区器的工作机制：Hash和Range两条路线

Spark的分区器主要有两个实现：HashPartitioner和RangePartitioner。

HashPartitioner的计算逻辑很简单，就是取分区键的hash值，再对分区数取模，算出这条记录该进哪个分区。公式大概是这样的：

partitionId = nonNegativeMod(key.hashCode, numPartitions)

这个方案的优点是快，缺点是如果多个不同的key hash之后取模结果相同，它们就会挤到同一个分区里。更麻烦的是，如果分区键本身有倾斜，那hash取模之后倾斜基本原样保留，满载的分区继续满载。

RangePartitioner主要用于sortByKey这类需要全局有序的场景。它会先对key做采样，估算出各个key的大致分布区间，再按区间把数据切成有序的段，保证某个区间内的key都进同一个分区，而且分区之间有大小顺序。整体数据分布通常更均衡，但采样和排序的开销更高。

2.3 执行计划里的Exchange节点和AQE的影响

在Spark的物理执行计划里，repartition会体现为一个Exchange节点。用explain查看执行计划时，能看到类似这样的描述：

Exchange hashpartitioning(business_key#123, 200)

这行信息非常有用。它直接告诉你两件事：用哪个字段做了hash分区，目标分区数是多少。排查性能问题时，第一步肯定是扫一遍执行计划，看看每个Exchange的分区键和分区数是不是符合预期。

还有一个必须提到的机制是自适应查询执行（AQE，Adaptive Query Execution）。开启spark.sql.adaptive.enabled之后，Spark会在shuffle阶段根据实际数据分布动态调整后续的分区策略。比如spark.sql.adaptive.coalescePartitions.enabled会自动把数据量太小的相邻分区合并，减少task数量和调度开销。AQE能在一定程度上缓解分区数设置不合理的问题，但它只能优化已有的分区策略，没法解决分区键选错带来的倾斜问题。

我自己排查任务的时候有个习惯：先看执行计划里每个Exchange的hashpartitioning字段，确认分区键对不对、分区数合不合理；再点开Spark UI里对应的stage，看task级别的数据分布是否均匀。这两步能过滤掉一大半的分区问题。

3. repartition和coalesce怎么选：一个决定性能的分支

很多入门资料会把repartition和coalesce放在一起讲，说“一个是增加分区，一个是减少分区”。这个说法没错，但不完整，它忽略了两者最本质的差异——是否触发shuffle。这个差异直接决定了两个API在性能上的数量级差距。

3.1 源码和行为上的核心差异

repartition的实现其实就是执行了一次coalesce，但强制指定了shuffle为true。

def repartition(numPartitions: Int): DataFrame = { coalesce(numPartitions, shuffle = true) }

coalesce的shuffle参数默认是false，也就是说coalesce(numPartitions)默认不会打散数据重排，它只是把多个现有分区合并成更少的分区。由于没有shuffle，这个操作非常轻量，几乎不产生网络和磁盘开销。

coalesce的实现还有一个特点：当目标是减少分区数时，它尽量在一个stage内部直接合并相邻分区，无需跨节点移动数据。所以如果你只是想把分区数从3000降到200，coalesce(200)通常比repartition(200)快一个数量级。

但coalesce也有硬限制：它不能在减少分区的同时保证数据均匀分布。因为它只做合并，不做重新分配。如果现有分区本来就倾斜，合并之后倾斜大概率还会存在。

3.2 选型规则：什么场景用哪个

实际项目里，我习惯按下面的规则来选：

这个表看起来简单，但实际工作中容易踩的坑恰恰是在“增大分区”和“减小分区”之外的那些灰色场景里。比如需求是“让下游并行度更高一些”，有人会直接repartition(500)，但根本没考虑目标分区数要不要跟任务的并行资源匹配。分区数设成500，如果executor核数只有100，那shuffle读阶段也只会起100个并发task，剩下400个分区排队处理，并行度优势完全发挥不出来。

3.3 代码示例和参数细节

看一个实际例子。假设拿到一个DataFrame，总大小在50GB左右，executor总核数是120。想让最终写入前每个文件大小控制在256MB附近，可以先估算目标分区数：

target_partitions = 50 * 1024 * 1024 * 1024 / (256 * 1024 * 1024) # 结果约 200 个分区

这个估算思路比拍脑袋定数字可靠得多。分区数跟文件数量直接相关，每个分区在写出时大概率对应一个文件。如果目标分区数是200，每个分区大约承载256MB数据，再配合spark.sql.files.maxRecordsPerFile之类的限制，文件大小会比较均匀。

另外要注意，DataFrame的repartition操作是lazy的，它不会立刻执行shuffle，而是等后面的action触发时才真正执行。所以不要在一个链路里反复repartition，最后真正生效的只有最后一次。中间的repartition只会白白增加计划复杂度，甚至让执行计划里的Exchange节点变多，增加优化器的负担。

4. 真实调优案例：数据倾斜下的分区键重设计与加盐方案

分区问题最典型、最伤人的表现就是数据倾斜。我在某数据分析链路里碰到过一个非常典型的场景：两张事实表join，join key是用户ID，但用户ID的分布极端不均，头部几个用户贡献了绝大部分记录数。这样的数据直接按用户ID分区，再做join，那几个头部分区必然成为瓶颈。

4.1 问题现象的完整描述

任务运行的时候，Spark UI里能看到某个stage的task处理时间出现了极端分化。几十个task在10秒内完成，两三个task需要跑十分钟。拖慢的不只是这个stage，后面依赖它的stage全部延迟，整个作业的调度资源也被这几个“长尾”task长期占用，同集群的其他任务跟着受影响。

更麻烦的是，这个问题不是靠增加资源能缓解的。哪怕把executor加到原来的三倍，倾斜的那几个分区由于数据量过大，单task处理时间仍然很长。加机器只能让空转的task变多，瓶颈反而更突出。

4.2 加盐方案的完整实现

针对这种由单key倾斜导致的join性能问题，业界最常用的做法是“加盐”——给倾斜的key打散成一个随机前缀，让原本集中在一个分区的数据被打散到多个分区里去。

具体分三步：

第一步，识别倾斜key。统计每个key的记录数，找到占比异常高的那些key。可以先用SQL做一次group by，按记录数排序，把头部若干key捞出来。

skew_keys = df.groupBy("user_id").count().orderBy("count", ascending=False).limit(10)

第二步，对倾斜key加盐，再和事实表加盐后join。思路是：把大表里倾斜key的每条记录，复制成N份，每份加上0到N-1的随机后缀；小表（或者维表）里对应key的数据也做同样的加盐操作，但复制方式要保证能与大表Join上。例如大表加盐后key变成user_id + "_" + salt，小表也要生成同样粒度的user_id + "_" + salt组合。

from pyspark.sql import functions as F # 大表：倾斜key加盐 df_large_skew = df_large.filter(F.col("user_id").isin(skew_keys)) \ .withColumn("salt", F.rand() * 10) \ .withColumn("join_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt").cast("int"))) # 非倾斜key保持原样 df_large_normal = df_large.filter(~F.col("user_id").isin(skew_keys)) \ .withColumn("join_key", F.col("user_id")) df_large_processed = df_large_skew.unionByName(df_large_normal) # 小表/维表：同样加盐，但需要把每个key复制N份 df_dim_skew = df_dim.filter(F.col("user_id").isin(skew_keys)) \ .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(10)]))) \ .withColumn("join_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt"))) df_dim_normal = df_dim.filter(~F.col("user_id").isin(skew_keys)) \ .withColumn("join_key", F.col("user_id")) df_dim_processed = df_dim_skew.unionByName(df_dim_normal)

第三步，join后按业务口径聚合或去重。由于大表加盐时每条记录都复制成了N份，join结果里会出现重复记录，需要按原始业务key和业务字段去重，或者直接按需要的聚合维度重新聚合。

result = df_large_processed.join(df_dim_processed, "join_key", "left") \ .drop("join_key", "salt") \ .dropDuplicates(["user_id", "record_id"])

4.3 为什么加盐能解决问题：分区数倍增的数学原理

加盐的本质是人为改变分区键的基数。原来只有一个user_id，现在有user_id + salt这个组合，基数扩大了N倍。配合repartition时指定的分区数，原本会落进同一个分区的数据，被打散到N个不同的分区中。每个分区承载的数据量骤降，单task处理时间大幅缩短。

当时这个任务加盐之后，倾斜最严重的那几个key被拆到了10个分区里去处理，整个join stage的耗时从原来的二十多分钟降到了四分钟左右。更重要的是，task之间的处理时间方差大幅缩小，集群资源利用率提升明显。

要注意的是，加盐方案不是万能的。如果倾斜key不止一个，或者倾斜程度极其极端，需要更精细的方案，比如CBO（基于代价的优化器）统计、手动拆分处理链路等。但90%的常规倾斜场景，加盐是最快见效的手段。

4.4 分区键重设计的一般原则

从这个案例可以提炼出分区键选择的通用原则：分区键的选择要同时考虑基数和分布均匀性。基数过低（比如只有几个枚举值）的字段不适合做分区键；分布严重不均的字段更不适合。实践中比较稳妥的做法是选择基数适中、分布相对均匀的字段，或者干脆不按业务字段分区，只做纯数量上的repartition(目标分区数)，让Spark按记录数大致切分。

如果业务上确实需要按某字段分区来优化后续join，可以配合AQE的spark.sql.adaptive.skewJoin.enabled参数，让Spark自动检测并处理join侧的倾斜。先让优化器干它能干的活，不够再手动加盐，这是最省力的路径。

5. 分区调整后的隐藏雷区：小文件、动态覆盖与检查清单

分区数调整完成、任务跑得快了，并不意味着万事大吉。我在实际项目里见过不少“调完分区反而出问题”的情况，Root cause往往不在重新分区本身，而在调整之后没注意那些隐藏的连锁反应。

5.1 小文件问题：分区数和文件数量的直接关联

分区调整最容易被忽略的连锁反应是“写文件时的小文件爆炸”。DataFrame写出时，每个分区通常会产生至少一个数据文件。如果做了一次repartition(5000)再写表，且每个分区里的数据只有几MB，落地之后就是几千个小文件。小文件会带来很高的NameNode元数据压力，后续对这张表做查询时，scan阶段要打开大量文件，产生大量的task调度开销，性能全线恶化。

所以调整分区数之前，一定要先想清楚这个DataFrame接下来要干嘛。如果是要写HDFS表，分区数的量级应该根据目标文件大小反推；如果只是中间环节的临时处理，那么后续还要再一次coalesce或repartition来压缩分区数，避免把小文件问题带到下游。

对比一下两个操作的区别：

5.2 repartition和partitionBy的区别：容易混淆的两个概念

很多新手会把DataFrame的repartition和写表时的partitionBy混为一谈。partitionBy是存储层的概念，它决定数据在文件系统里怎么按目录组织，比如按日期、按地区建目录。repartition是计算层的概念，它决定数据在集群内存和磁盘上怎么分布。

两者的关系是：写表时可以先用repartition整理数据分布，再用partitionBy控制存储目录结构。但如果partitionBy的字段和repartition的字段不一致，Spark在写入时可能还会额外做一次shuffle，防止同一个目录下的数据在多个task里并发写出产生冲突。

实际项目里遇到“写了表但目录下文件特别多”的问题，多半是这两层没配合好。正确的做法是先想清楚存储目录的层级和粒度，再反推计算层需要多少个分区。比如目标目录粒度是“天”，天均数据量是300GB，目标文件大小是256MB，那每天目录下的文件数大约是1200个，对应每个分区承载256MB左右的数据，分区数就可以按这个量级来定。

5.3 调整分区之后的完整检查清单

结尾

最后分享一个我个人的排查习惯。每次需要调整DataFrame分区之前，我会先在草稿纸上回答三个问题：当前的数据总量和期望文件大小分别是多少，目标分区数是否匹配；有没有字段分布不均匀的风险，是否需要先做倾斜检查；这次调整之后数据接下来要去哪，是否需要再压缩分区。三个问题答完了，再动手写代码。

分区这件事，说复杂很复杂，底层牵扯shuffle、分区器、AQE一大堆机制；说简单也简单，核心就一条：让数据在正确的位置、以合理的粒度进行处理。真正决定任务性能的，往往不是某个单一参数，而是整个链路里所有分区操作叠加后的最终效果。每次重新分区之前多问一句“为什么”，能省下后续无数的排查时间。