企业官网建设流程全解析

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时间哗哗地涨。最极端的情况是出现OOM，Executor直接挂掉，然后Spark开始重算那个分区的数据，任务失败重试，又失败又重试，整个Job就这么被拖死。

1.3 分区太多：矫枉过正

分区过多同样有麻烦。每个Task调度是有开销的，比如序列化、任务分发、结果收集，这些都需要时间和内存。如果数据量不大却设了上万的分区，每个Task只处理几KB数据，那么整个Job大部分时间都耗在调度管理上，真正的计算反而没多少。

写文件的时候，分区过多还会直接导致小文件问题。每个Task如果最后各自写文件，10000个分区就产生10000个小文件。HDFS上的NameNode内存扛不住海量小文件的元数据，下游任务读这些小文件时也慢得出奇，因为打开每个文件都有额外开销。

所以分区数并不是越大越好，它需要和数据量、集群资源匹配。我自己的经验习惯是，把目标分区大小控制在128MB到256MB之间，这也是Spark读文件的默认块大小附近的值。有一个比较实际的经验：如果一个Stage需要shuffle，分区数先按总数据量 / 目标分区大小来估算，再结合集群并行度往上取整。

2. 重新分区的核心操作：repartition、coalesce 与 partitionBy

2.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以及同样多的分区数进行repartition，join时就能变成结构化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/province=xxx/city=yyy/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 BY：

INSERT 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.enabled=true，在Spark 3.x中会自动合并shuffle后过小的分区，这个功能对减少小文件非常有帮助，建议生产环境直接开启。

3.3 join前按join key重新分区

大表join是另一个常见场景。说一个实际优化效果明显的例子：一张用户维表（几百万条）和一张用户行为大表（几亿条）按user_id关联。默认情况下，Spark需要将大表按user_id做一次shuffle，把相同user_id的数据发到同一个节点上才能Join。这个shuffle是必须的，但你可以控制它最终分成几个分区以及分区是否均衡。

操作上就是提前对参与Join的两个DataFrame做repartition：

val 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。

操作步骤如下：

1. 从两张原始表中分别过滤掉热点key，得到非热点数据，直接用普通join。
2. 单独取出热点key的大表数据，对这些数据加盐为多个重新分区。
3. 取出对应热点key的小表数据，复制多份并加相同盐值，使其与大表加盐后的key对齐。
4. join第2步和第3步的结果。
5. 把两部分的join结果union起来。

这个方式比全面加盐效率高，因为它只对热点key做处理，非热点数据走常规路径，不会给所有key都增加一个维度。但代码复杂度也上来了，需要多写不少逻辑。如果倾斜并不是特别严重，我觉得先试两阶段聚合就够了，除非到了无法容忍的程度再上这套组合拳。

5. 排查分区问题与小文件治理实践

5.1 怎么查看当前分区数

排查问题前总得知道数据现在被分成了多少个分区。DataFrame提供了直接获取分区数的方法：

df.rdd.getNumPartitions

Python也是一样的调用方式：

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动态分区合并和自适应执行。

这里分享一组我现在生产环境常用的小文件治理参数组合：

这套参数组合配好后，最常见的效果就是：写出的结果文件数量大幅减少，并且每个文件大小保持在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在读取阶段就做分区裁剪，这比任何重新分区方案都更省成本。分区优化这件事，永远不是从计算阶段才开始的。