Spark DataFrame重新分区这件事表面上看不就是调个repartition(n)或coalesce(n)嘛但真跑到生产环境里分区数设不对、分区方式选错轻则任务慢几倍重则直接OOM或者写出一堆小文件把你整个数仓都搞崩。我之前就被折腾过一整个晚上任务调度看起来正常但Spark UI上那个Stage进度条就是卡在99%点开单个Task一看某几个Executor在疯狂处理几千万条数据其余全部闲置——数据倾斜加分区不合理典型的案例。后来把重新分区这套机制吃透了才真正体会到分区不只是“设置一个并行度”它直接影响shuffle成本、文件落地方式、内存消耗和故障恢复速度。这篇文章我就围绕DataFrame的重新分区把repartition、coalesce、partitionBy这几个东西讲透再结合真实场景里的分区数选择、数据倾斜优化、小文件治理这些硬骨头把排查思路和参数调优经验一并写出来。特别适合每天跟Spark SQL、批任务打交道的朋友也适合刚入手Spark但总被分区概念绕晕的学习者。1. 分区机制到底影响什么1.1 一个分区就是一个并行度Spark里的分区全称是RDD/DataFrame的分区Partition它是数据在集群中被划分成的一个个独立单元。每个分区对应一个Task一个Task又会被分配到某个Executor的一个Core上执行。换句话说分区数量就是你的任务的并行度上限。打个比方就像快递分拣线上有一堆包裹要处理。每个分拣员就是一个Task包裹就是数据。分区数就是分拣员的个数每个分拣员手里拿着一个筐筐里的包裹只归他处理。如果你只有1个分拣员即使有100台设备在那空着也只能一个接一个处理那效率肯定低得吓人。这个设计背后藏着Spark的两个核心机制Task是Spark执行的最小单位一个分区的数据从头到尾都由一个Task处理。分区的边界决定了数据本地性Task会尽量调度到数据所在的节点上减少网络传输。所以分区数设大了并行度就高资源利用率也上来设小了CPU和内存都闲着。很多任务跑得慢先别怀疑代码逻辑有问题先看一眼Spark UI上Executor的数量和每个Stage的Task数量是不是合理。1.2 分区太少资源白白浪费分区太少的后果比较直观。假设你有一个Executors的集群每个Executor分配了2核那整个集群能同时跑的Task总数就是 Executors数 × 2。如果你的数据被分成了80个分区而集群能跑100个Task那就意味着有20个核在空转任务总体耗时至少不是最优的。更严重的是如果只有5个分区每个Task就得处理很大量的数据。数据在内存里放不下就要溢写到磁盘然后在溢写的过程中反复排序和合并磁盘IO和GC时间哗哗地涨。最极端的情况是出现OOMExecutor直接挂掉然后Spark开始重算那个分区的数据任务失败重试又失败又重试整个Job就这么被拖死。1.3 分区太多矫枉过正分区过多同样有麻烦。每个Task调度是有开销的比如序列化、任务分发、结果收集这些都需要时间和内存。如果数据量不大却设了上万的分区每个Task只处理几KB数据那么整个Job大部分时间都耗在调度管理上真正的计算反而没多少。写文件的时候分区过多还会直接导致小文件问题。每个Task如果最后各自写文件10000个分区就产生10000个小文件。HDFS上的NameNode内存扛不住海量小文件的元数据下游任务读这些小文件时也慢得出奇因为打开每个文件都有额外开销。所以分区数并不是越大越好它需要和数据量、集群资源匹配。我自己的经验习惯是把目标分区大小控制在128MB到256MB之间这也是Spark读文件的默认块大小附近的值。有一个比较实际的经验如果一个Stage需要shuffle分区数先按总数据量 / 目标分区大小来估算再结合集群并行度往上取整。2. 重新分区的核心操作repartition、coalesce 与 partitionBy2.1 repartition全量shuffle成本最高但最稳定repartition是Spark里最常用的重新分区方法。它会把数据完全打乱然后按照新的分区数重新分配。因为涉及全量网络传输它是一个宽依赖操作代价非常大但好处是通常能保证数据相对均匀地分布到各个分区。它有两种典型用法// 只指定分区数数据按轮询方式均衡分发 val df2 df.repartition(200) // 按某列进行哈希分区相同key的数据进入同一分区 val df3 df.repartition(200, col(user_id))第一种更适合调整并行度比如从上游读了一个没有经过shuffle的DataFrame分区数偏少想增大分区提高后续计算并行度。第二种适合做join缓存如果两张表之后要按user_id做join提前按这个key进行相同分区数的重分区join时就能减少或避免后续的shuffle。但实际上Spark的join优化器有自己的shuffle机制如果两张表都分别按同一个key以及同样多的分区数进行repartitionjoin时就能变成结构化join省掉一次Exchange。使用repartition时要注意它不是廉价的。如果频繁调用每个Job都会多出一次全量shuffle。尤其数据量很大的时候一次repartition可能要跑十几分钟所以能少用就少用。但当你需要把数据从极度倾斜的状态中“洗匀”时repartition又是最可靠的选择因为coalesce解决不了倾斜问题。2.2 coalesce窄依赖减少分区的最优选coalesce是用来缩减分区数的这个方法缩写的含义就是“合并”。它会尽量在map端完成把多个分区合成一个能避免shuffle就避免shuffle所以成本比repartition低得多。val df2 df.coalesce(5)注意一个核心行为coalesce如果目标是减少分区它不会发生真正的数据重洗只是把某些分区直接拼接起来。这带来一个隐藏问题——数据分布不均衡。比如原本有100个分区每个分区处理的数据量差不多你现在coalesce(10)它往往是把相邻的10个分区合并成1个合并前每个分区的数据量如果参差不齐合并后的数据量也会非常不均匀。所以coalesce适合的场景是你已经做了一个shuffle操作比如groupBy或者repartition数据在分区中本来就比较均衡只是最后要落地成数量更少的文件时用它减少文件数量。最典型的例子是写数据前df.coalesce(1)来输出单个文件。如果直接从源文件读取后的DataFrame做coalesce这个操作基本没有意义因为你不知道原分区的数据分布合出来的分区可能还是倾斜的。反过来说coalesce有一个限制它不能增大分区数。因为如果要从100个分区变成200个分区就必须拆数据那就需要shuffle所以那样做只能使用repartition。2.3 partitionBy写文件时的目录分区partitionBy是DataFrameWriter的一个参数很多人把它和内存里的分区操作搞混。它和repartition、coalesce的关注点完全不同。repartition和coalesce影响的是数据在内存中如何被切分partitionBy影响的是数据写到文件系统时按什么字段建立目录结构。df.write .mode(overwrite) .partitionBy(province, city) .parquet(/data/user_info)执行完上面这段代码数据在HDFS上的目录结构大概是/data/user_info/provincexxx/cityyyy/xxx.parquet。这样做的好处是下游查询时Spark引擎可以根据分区字段直接做分区裁剪跳过不需要的目录查询效率大幅提升这也是Hive数仓最常见的建模方式。但partitionBy和内存分区之间有一个联动关系写文件时每个内存分区会对应生成一个或者多个文件。如果你的DataFrame有200个内存分区而partitionBy只有一个分区键那么每个分区目录下最多可能出现200个文件这就会造成文件碎片化。解决办法通常是先按分区键做repartition让同一个输出目录的数据集中在少数几个分区里再用partitionBy写。这也是结合两者最常用的组合操作。3. 实际场景下的重新分区实现3.1 用SQL写Hive表时控制shuffle分区在Spark SQL里执行INSERT OVERWRITE TABLE xxx SELECT ...这类操作时如果SELECT中带了GROUP BY、DISTINCT、JOIN就会触发shuffle而shuffle后的分区数量由参数spark.sql.shuffle.partitions控制默认值是200。很多人会忽略这个参数。如果你的任务处理的数据量不大200个分区其实偏多每个Task写出来的文件都很小造成小文件堆积。而如果数据量特别大200个分区又可能不够用每个Task分配到的数据过多内存压力陡增。我通常的做法是先估算这个Stage的输入数据规模然后用总数据量 / 128MB计算理想分区数再根据实际执行情况做微调。比如处理一份大约50GB的中间结果那么理想分区数就是50GB / 128MB ≈ 400个分区我就把spark.sql.shuffle.partitions设成400左右。处理几十GB以内的数据也可以保守一点先观察默认的200再看Spark UI中Task的Shuffle Read大小如果单个Task读了几百MB甚至更多说明分区太少就调大如果单个Task只读了几MB说明分区太多就调小。3.2 动态分区插入时的分区优化往Hive表里做动态分区插入是很常见的数仓场景。比如把每天的日志表按日期字段动态分区写入结果表结果表的分区字段是dt每天一个分区。这里有一个很容易踩的坑。如果SELECT结果没有经过合理分区直接动态插入Spark会根据spark.sql.shuffle.partitions决定中间的分区数量。假设这个大表有几百亿条数据你的shuffle分区设了100个那么落到每个动态分区下的文件数就比较多每个文件大小也参差不齐最终HDFS上会积压大量小文件。为了控制文件数量一种做法是写SQL时使用DISTRIBUTE BYINSERT OVERWRITE TABLE dwd_user_log PARTITION (dt) SELECT ..., dt FROM ods_user_log DISTRIBUTE BY dt;这里的DISTRIBUTE BY等价于repartition它保证相同dt的数据进入同一个内存分区这样写出到对应分区目录下的文件就非常集中。如果你希望每个分区内的文件更均匀还能进一步DISTRIBUTE BY dt, cast(rand() * N as int)相当于把每个分区再切成N块兼顾均匀和文件数量。另外通过spark.sql.adaptive.enabled配合spark.sql.adaptive.coalescePartitions.enabledtrue在Spark 3.x中会自动合并shuffle后过小的分区这个功能对减少小文件非常有帮助建议生产环境直接开启。3.3 join前按join key重新分区大表join是另一个常见场景。说一个实际优化效果明显的例子一张用户维表几百万条和一张用户行为大表几亿条按user_id关联。默认情况下Spark需要将大表按user_id做一次shuffle把相同user_id的数据发到同一个节点上才能Join。这个shuffle是必须的但你可以控制它最终分成几个分区以及分区是否均衡。操作上就是提前对参与Join的两个DataFrame做repartitionval left leftDf.repartition(400, col(user_id)) val right rightDf.repartition(400, col(user_id)) val joined left.join(right, Seq(user_id), left)这么做的意义有两个。第一如果之后还要对join结果再次聚合或写入提前重分区能让后续操作少一次shuffle第二如果某些user_id的数据量特别大比如某个热门用户手动重分区后数据分布仍然可能倾斜这时候配合加盐策略可以绕开热点。但别为了“优化”盲目加这种操作。如果两张表里的高频key不多或者数据整体较均衡Spark自己做的动态shuffle就已经够了。手动repartition前建议先在Spark UI上看当前Stage的Shuffle Read量时长是否异常再去决定要不要干预。3.4 导出单文件时的取舍还有一种很常见的需求把DataFrame处理结果导出成单个CSV或Excel交给业务人员。导出单文件时你可以写df.coalesce(1)或df.repartition(1)。区别在于coalesce(1)不会触发全量shuffle只把所有分区数据合并到一个Task性能上通常更快而repartition(1)会把所有数据打散再集中到一个分区过程更重但结果一般也是均衡的。对于导出单文件这个场景coalesce(1)就足够。但这个操作不能乱用。如果这个DataFrame特别大把所有数据汇聚到一个分区去写文件会带来两个问题单Task处理数据量过大内存溢出风险极高。单Task写一个文件文件持久化时耗时长而且网络IO集中在单个节点上很容易成为瓶颈。所以我通常建议如果不是业务硬性要求不要导出单文件。如果必须导出尽量控制数据量在1GB以内或者先在集群里把结果写成分区文件再用其他方式合并下载。4. 重新分区与数据倾斜的博弈4.1 如何快速定位分区数据倾斜数据倾斜和重新分区经常同时出现。在Spark UI里多个Task中有一个Task长期运行不结束其他Task早就完成了这种明显的长尾现象就是倾斜的典型信号。另外一个信号是Shuffle Read时某个Task拉取了远超平均水平的数据量比如其他Task都只读50MB而有一个Task读了几GB。为什么会倾斜核心还是分区时key分布不均匀。按某个字段做哈希分区后如果某个key占的数据量特别大那这个key所在的分区数据就特别多。常见倾斜原因包括用户ID字段为空、城市字段集中于某几个热门值、状态字段只有少数几个取值。如果使用repartition(col)按这些字段分区倾斜基本很难避免。解决思路分两种一种是绕开另一种是打散后重新聚合。4.2 两阶段聚合加盐打散热点如果是做groupBy聚合而导致倾斜两阶段聚合是最对症的解法思路就是加盐。加盐的意思是在原来Key的基础上拼接一个随机数前缀让相同Key的数据被随机分散开完成第一轮局部聚合然后再去掉随机前缀做第二轮聚合。代码上看大致是这样val salted df.withColumn(salt, (rand * 10).cast(int)) .withColumn(salted_key, concat(col(key), lit(_), col(salt))) val partial salted.groupBy(salted_key).agg(sum(value).as(partial_sum)) val result partial.withColumn(key, split(col(salted_key), _).getItem(0)) .groupBy(key).agg(sum(partial_sum).as(total_sum))这里rand() * 10中的10就是盐的份数代表把每个热点key最多打散成10份。这个值不是越大越好。加盐数量越大第一阶段的中间分区越多内存和网络开销也越大。如果倾斜的key不多10~50之间通常够用。加盐后热点key被分散到不同分区和不同Task局部聚合时压力被分解等到二次聚合时每个Task需要处理的数据量已经大幅下降。但这个方案只适用于聚合不适用于join。Join场景下你给大表加盐小表也得做同样的扩展才能匹配否则Join不上。小表扩展的方式是复制多份这种方案也有用但要注意小表不能太小也不能太大得控制在一个能广播或者能被join承担的范围内。4.3 热点key单独拆分处理还有一个更精细的方案先把热点key识别出来单独处理。比如根据某个历史统计某几个商品ID或用户ID贡献了绝大部分数据那么在join前先分别统计两个表里key的频次找出top 10的热点key。操作步骤如下从两张原始表中分别过滤掉热点key得到非热点数据直接用普通join。单独取出热点key的大表数据对这些数据加盐为多个重新分区。取出对应热点key的小表数据复制多份并加相同盐值使其与大表加盐后的key对齐。join第2步和第3步的结果。把两部分的join结果union起来。这个方式比全面加盐效率高因为它只对热点key做处理非热点数据走常规路径不会给所有key都增加一个维度。但代码复杂度也上来了需要多写不少逻辑。如果倾斜并不是特别严重我觉得先试两阶段聚合就够了除非到了无法容忍的程度再上这套组合拳。5. 排查分区问题与小文件治理实践5.1 怎么查看当前分区数排查问题前总得知道数据现在被分成了多少个分区。DataFrame提供了直接获取分区数的方法df.rdd.getNumPartitionsPython也是一样的调用方式df.rdd.getNumPartitions()想查某个字段在各分区的分布情况可以用Spark UI的Stage详情页看也可以写个临时统计任务df.groupBy(key).count().orderBy(desc(count)).show()但这只能估出key的总体分布没法直接看出每个Task处理了多少数据。更直接的是Spark UI里某个Stage的Task列表表格有Shuffle Read Size / Shuffle Write Size列那列数字一排序倾斜的Task一眼就露出来了。生产环境我见过最典型的坑就是代码里写了一个df.repartition(1000)然后下面接了一个写出操作而数据才几百MB。结果就是1000个Task每个写一个文件下游读这1000个小文件慢得像蜗牛爬。排查步骤很简单看数的分区大小问题就浮出来了。5.2 小文件治理的参数组合小文件治理是Spark作业绕不开的主题。除了减少不必要的repartition更重要的是合理利用Spark动态分区合并和自适应执行。这里分享一组我现在生产环境常用的小文件治理参数组合参数推荐值作用说明spark.sql.adaptive.enabledtrue启用Spark自适应执行框架spark.sql.adaptive.coalescePartitions.enabledtrue自动合并shuffle后过小的分区spark.sql.adaptive.advisoryPartitionSizeInBytes64MB合并后的目标分区大小建议值spark.sql.adaptive.coalescePartitions.minPartitionNum与集群并行度匹配合并后最小分区数避免压得过低spark.sql.shuffle.partitions根据数据量估算SQL shuffle分区基数spark.sql.maxPartitionBytes128MB控制读文件时的分区大小这套参数组合配好后最常见的效果就是写出的结果文件数量大幅减少并且每个文件大小保持在64MB左右而不是以前那种几百字节一个的小文件。特别是调度任务多、定时任务频繁跑的项目这个调整后任务稳定性提升非常明显。还需要注意一点历史已经生成的存量小文件参数配得再好也救不了。对这种存量数据需要用REWRITE或OPTIMIZE的方式重新整理比如读一次源目录再repartition按目标大小写出覆盖。这样才能真正把小文件数量降下来。5.3 用执行计划判断分区策略是否生效调优时怎么确认自己写的repartition有没有在计划里生效最直接的办法就是看执行计划。df.groupBy(key).agg(sum(value)) .explain(detailed)输出里会包含Exchange节点里面写明使用的是hashpartitioning(key, 200)还是RangeCoalesce之类的逻辑。如果看到计划里的分区数和预期不符就可以回头检查是不是参数没生效或者代码路径没走到。日常排查时养成看执行计划的习惯非常有用。很多同事找我排查Spark任务慢的问题第一反应就是改参数很少有先看explain的。执行计划是一个真正权威的“诊断报告”它会把每个阶段的分区策略、shuffle位置、数据扫描范围都标得清清楚楚。看懂它基本就掌握了寻找任务优化切口的主动权。一点个人经验分区这块我踩过的最深的一个坑就是太依赖网上搜索来的固定参数值。分区数、分区策略都不是独立存在的它们和数据规模、集群资源、任务类型强相关。我现在每次拿到一个Spark任务习惯先看一眼输入数据量和执行计划再估算目标分区大小跑一轮后回到Spark UI确认Task执行情况有问题再针对性地调整。与其死记硬背参数不如掌握这套“估算 — 观察 — 调优”的方法越用越顺手。最后分享一个小技巧如果任务里有时间段范围查询比如数仓里查询某一天的数据可以优先考虑直接在SQL里把分区字段写进WHERE条件让Spark在读取阶段就做分区裁剪这比任何重新分区方案都更省成本。分区优化这件事永远不是从计算阶段才开始的。