上个月排查一个跑批任务变慢的问题最后定位到的根因让整个组都挺意外数据量其实没怎么涨但作业从十几分钟拖到了四十分钟。顺着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的分区键和分区数是不是符合预期。还有一个必须提到的机制是自适应查询执行AQEAdaptive 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降到200coalesce(200)通常比repartition(200)快一个数量级。但coalesce也有硬限制它不能在减少分区的同时保证数据均匀分布。因为它只做合并不做重新分配。如果现有分区本来就倾斜合并之后倾斜大概率还会存在。3.2 选型规则什么场景用哪个实际项目里我习惯按下面的规则来选场景推荐API原因增大分区数、提升并行度repartition必须打散重排才能重新平衡减小分区数、减少小文件coalesce无shuffle开销小按某个业务字段重新组织数据以备后续joinrepartition(分区键)让相同key的数据落在同一分区减少join时的shuffle过滤了大量数据后压缩分区coalesce数据量变少直接合并即可结果均匀分布后再写入下游repartition(分区数)保证各分区数据量接近这个表看起来简单但实际工作中容易踩的坑恰恰是在“增大分区”和“减小分区”之外的那些灰色场景里。比如需求是“让下游并行度更高一些”有人会直接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. 真实调优案例数据倾斜下的分区键重设计与加盐方案分区问题最典型、最伤人的表现就是数据倾斜。我在某数据分析链路里碰到过一个非常典型的场景两张事实表joinjoin 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, ascendingFalse).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来压缩分区数避免把小文件问题带到下游。对比一下两个操作的区别操作对输出文件数量的影响shuffle适用阶段repartition(n)输出约n个文件全量shuffle需要重新分布数据时coalesce(n)输出不超过原分区数的文件无shuffle仅合并压缩分区时partitionBy(字段)按字段值生成目录和文件可能触发额外shuffle按业务字段组织存储时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一大堆机制说简单也简单核心就一条让数据在正确的位置、以合理的粒度进行处理。真正决定任务性能的往往不是某个单一参数而是整个链路里所有分区操作叠加后的最终效果。每次重新分区之前多问一句“为什么”能省下后续无数的排查时间。