shuffle 并行度是什么
Spark SQL 里 spark.sql.shuffle.partitions 决定 shuffle 之后产生多少个分区，也就是 join、group by 这类算子下游的并行度，默认值是 200。分区偏小，单个 task 要处理的数据过多，容易内存溢出并放大倾斜；分区偏大，task 数与输出文件数同步膨胀，调度开销和小文件问题接踵而来。经验做法是先估算总 shuffle 数据量，再按单个分区 100-200MB 反推分区数，剩下的交给 AQE 在运行时合并或拆分。
基本信息
| 核心参数 | spark.sql.shuffle.partitions |
|---|---|
| 默认值 | 200（Spark 3.x 仍为 200） |
| 经验值 | 单个 shuffle 分区处理 100-200MB 数据 |
| 估算方式 | 总 shuffle 数据量除以 200MB 后取整 |
| 运行时合并 | 开启 AQE 后由 coalescePartitions 自动合并小分区 |
| 读入并行度 | 由文件切分参数决定，与 shuffle 分区数是两回事 |
| 写入前收敛 | repartition 会走 shuffle，coalesce 只减少分区不洗牌 |
典型使用场景
- 大表 join 后单分区过大、长尾明显：提高 shuffle 分区数把数据摊开
- 小数据量作业仍按默认 200 分区跑：产生大量小文件，应调低或交给 AQE 合并
- 写入 Hive 或湖表前按分区列 repartition，让输出文件数可控
- 读取一批小文件时并行 task 过多：通过文件切分相关参数把多个小文件合并进一个读分区
- 任务在同一份数据上时快时慢：先看分区大小分布，再判断是倾斜还是并行度不足
并行度调参示例
并行度调参示例
spark.conf.set('spark.sql.shuffle.partitions', '400') # 按数据量估算：总量除以 200MB
spark.conf.set('spark.sql.adaptive.enabled', 'true') # 打开 AQE
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')
spark.conf.set('spark.sql.files.maxPartitionBytes', '134217728') # 读入阶段单分区 128MB
df = spark.sql("SELECT store_id, sum(pay_amt) AS amt FROM dwd_trade_order_di WHERE dt='2026-10-12' GROUP BY store_id")
df.repartition(64, 'store_id') # 写入前重分布，控制落盘文件数量
df.write.mode('overwrite').insertInto('dws_trade_store_day') # 目标表按 store_id 分桶/分区时更均匀
df.coalesce(8) # 只减少分区，不触发 shuffle，适合小结果集相关参数对照
| 参数 | 作用 | 典型取值 |
|---|---|---|
| spark.sql.shuffle.partitions | shuffle 后分区数 | 按单分区 100-200MB 估算 |
| spark.sql.adaptive.enabled | 是否开启自适应执行 | true |
| spark.sql.adaptive.coalescePartitions.enabled | 运行时合并小分区 | true |
| spark.sql.files.maxPartitionBytes | 读入阶段单个分区的数据量 | 128MB（默认） |
shuffle 分区数与读入并行度是两个不同的开关：前者只影响 join、group by 之后的阶段，后者由文件切分参数决定。调错对象时现象往往是「参数改了但任务没有任何变化」。
分区数调大并不总是更快。当每个分区只有几 MB 数据时，task 调度与文件打开开销会超过并行带来的收益，这也是 AQE 自动合并小分区被广泛开启的原因。