Spark 汇总作业越跑越碎：先把 shuffle 分区数从默认 200 改对

某制造企业的日终汇总作业最近越跑越不对劲：数据量没有明显增长，跑批时间却从十几分钟拖到半小时以上，下游一查输出目录，几百个几十 KB 的小文件躺在那里。开发第一反应是加资源、加并行度，结果越调越慢。真正的问题不在资源上，而在一个几乎没人动过的默认值上：Spark SQL 的 shuffle 分区数。

默认 200 分区，对谁都一样

只要 SQL 里出现聚合、去重或者 Join，Spark 就会做 shuffle，把数据按 Key 重新分发到下游任务。这个分发份数由参数 spark.sql.shuffle.partitions 控制，Apache Spark 官方性能调优文档给出的默认值是 200。也就是说，不管 shuffle 后的数据是 50MB 还是 500GB，默认都会切成 200 份。

对小作业来说，200 份就是 200 个任务，每个任务只处理几百 KB，还要各自启动、各自写一个输出文件，任务启动和文件提交的开销反而成了大头。对大作业来说，200 份又可能太少，单个分区动辄几个 GB，任务容易内存溢出或者频繁落盘，拖慢整个 Stage。同一个默认值，在两端都会出问题，这就是它最容易被忽视的地方。

定位时先不要急着改参数，先看两个地方。第一是 Spark UI 里 shuffle 阶段的任务数和每个任务的 Shuffle Read 大小：如果任务数正好是 200，而单个任务的读取量只有几百 KB，基本就是分区切碎了；如果单个任务的读取量达到 GB 级并且伴随大量 Spill，那就是分区切少了。第二是输出文件数：一个汇总作业每天写出 200 个小文件，下游再读这张表时，光是列文件、开文件的开销就已经很可观了。

先算一笔账，再定分区数

分区数没有放之四海而皆准的值，但有一个可直接套用的算账办法：用 shuffle 后的数据量除以目标分区大小。目标分区大小取 64MB 到 128MB 之间是一个常见的经验区间，前者对应 AQE 合并时的参考目标，后者对应读文件时的默认分片大小。假设某次汇总的 shuffle 数据量大约是 2GB，按 128MB 估算：

分区数 ≈ 2048MB ÷ 128MB = 16

也就是说，这类作业把分区数设为 16 到 32 之间，比默认 200 合理得多。估算不需要很精确，数量级对了就能去掉大部分浪费；真正精确的数字应该来自 Spark UI 里实测的 Shuffle Read 总量，而不是凭感觉拍一个数。

下面是一段可以直接跑的日终汇总示例，按班组汇总当天的产量和合格数。关键在建会话时就把分区数定下来，而不是等作业跑起来再想：

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("daily_output_summary")
    .config("spark.sql.shuffle.partitions", "32")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

summary = spark.sql("""
    SELECT
        workshop_id,
        team_id,
        prod_date,
        SUM(output_qty) AS total_output,
        SUM(qualified_qty) AS total_qualified
    FROM ods.daily_production
    WHERE prod_date = '2026-11-04'
    GROUP BY workshop_id, team_id, prod_date
""")

summary.write.mode("overwrite").parquet("/data/warehouse/dws/daily_output_summary")

如果用纯 SQL 提交，也可以直接在作业开头写两行配置，效果是一样的：

SET spark.sql.shuffle.partitions = 32;
SET spark.sql.adaptive.enabled = true;

这里的 32 不是标准答案，而是按上面那笔账算出来的起点：2GB 数据、128MB 目标分区大小，算出来是 16，再留一倍余量应对数据波动，取 32。数据量变了，起点也要跟着重算，照抄 32 同样会犯和照抄 200 一样的错误。

AQE 不是万能兜底，别指望它替你算账

Spark 3.0 之后引入的自适应查询执行 AQE，确实能在运行时合并过小的 shuffle 分区。官方文档给出的默认值是：spark.sql.adaptive.coalescePartitions.enabled 为 true，合并时的参考目标 spark.sql.adaptive.advisoryPartitionSizeInBytes 为 64MB。听起来只要开着 AQE，200 这个默认值就无所谓了，实际并非如此。

AQE 的合并是在 shuffle 完成之后、根据实际统计信息再调整，它能救小分区过多的问题，却救不了分区过少导致的大分区溢出：一个已经几 GB 的分区，在合并阶段只会被原样保留，不会被自动拆小。AQE 里确实有处理倾斜分区的拆分逻辑，但那针对的是个别 Key 倾斜造成的个别大分区，不是整体分区数定少了的情况。更重要的是，AQE 的合并行为还受其他开关影响，比如以并行度优先的模式会忽略 64MB 的目标大小，只按最小分区大小合并，合并效果和预期可能差很多。

所以正确的心态是：AQE 是保险，不是算账的替代品。分区数的起点仍然要自己按数据量算，AQE 只负责在实际数据量和预估有出入时做二次修正。把希望全押在 AQE 上，等于把本该在提交前定好的数，推迟到运行时去碰运气。

分区数定好了，输出文件数还要单独管

即使 shuffle 分区数算对了，输出文件数也不一定正好等于分区数。如果汇总结果本身只有几 MB，却写出 32 个文件，对下游来说仍然是小文件问题。这时不要回头再改 shuffle 分区数去凑文件数，两件事应该分开处理：shuffle 分区数管计算阶段的并行度，写出前用 coalesce 管最终文件数。

summary.coalesce(4).write.mode("overwrite").parquet("/data/warehouse/dws/daily_output_summary")

把 32 个分区在写出前合并成 4 个文件，计算阶段的并行度不受影响，下游读到的文件数也降下来了。这里的 4 同样是经验值，取决于结果集大小和下游的读取方式，结果集只有几 MB 时取个位数就够了。反过来也一样：不要为了少写几个文件而把 shuffle 分区数压到个位数，那会直接拖慢聚合和 Join 的计算阶段。

还有一个容易混淆的点：coalesce 不经过 shuffle，合并时只是把分区拼在一起，适合在写出前收敛文件数；而 repartition 会重新 shuffle，适合在数据倾斜或者分区分布不均时重排数据。写出前收文件用前者，需要重分布才用后者，用反了又会多一次无谓的 shuffle。

小结

Spark 汇总作业跑得碎、跑得慢，先看 spark.sql.shuffle.partitions 是不是还停在官方默认的 200。定位时看 Spark UI 里 shuffle 阶段的任务数和单任务读取量，定值时按 shuffle 数据量除以 64MB 到 128MB 的经验区间去算，2GB 左右的汇总作业从 32 起步再按实测调整；AQE 默认开启的分区合并、参考目标 64MB 只能做二次修正，救不了整体分区定少导致的大分区溢出。最后把计算分区数和输出文件数分开管：前者按数据量算，后者写出前用 coalesce 收敛。分区数算对了，任务数、溢写和小文件这三个症状往往会一起消失。