Shuffle 是什么
shuffle 是宽依赖算子在执行时把数据按 key 重新分发到不同节点的过程，group by、join、distinct、全局排序都会触发。它同时消耗磁盘 IO、网络带宽和序列化开销，还要经历一次完整的落盘与再读取，因此是批作业里最贵的一步。窄依赖算子（map、filter、union 这类逐个分区内部处理的算子）不会产生 shuffle，调优时先分清这一步是宽依赖还是窄依赖。
基本信息
| 触发条件 | 按 key 重分区：聚合、连接、去重、排序 |
|---|---|
| 主要代价 | 磁盘落盘与回读、网络传输、序列化与反序列化 |
| 常见优化 | 预聚合、广播小表、自适应合并小分区、加盐打散 |
| 适用引擎 | Spark、Flink 批模式、MapReduce 均有同类机制 |
| 观察入口 | Spark UI 的 stage 详情、Flink Web UI 的算子指标 |
Spark 相关配置
Spark 相关配置
spark.sql.shuffle.partitions=400 # 默认 200，按数据量调整
spark.sql.adaptive.enabled=true # 3.x 默认开启，自动合并小分区
spark.sql.adaptive.coalescePartitions.enabled=true # 允许 AQE 合并相邻小分区
spark.sql.adaptive.advisoryPartitionSizeInBytes=128m # 合并后的目标分区大小
spark.sql.autoBroadcastJoinThreshold=10485760 # 小表广播阈值（默认 10MB）
spark.reducer.maxSizeInFlight=48m # reduce 端单次拉取量，网络差时可调小典型使用场景
- 两张大表 join：先按连接键预聚合或过滤瘦身，再进入 shuffle，能同时减少传输量与计算量
- 分组统计：用 combineByKey、reduceByKey 这类接口做 map 端预聚合，把同 key 的局部结果先合并再发
- 小表 join 大表：改用广播 join，把小表复制到每个 executor，大表就不再重分布
- 数据倾斜：给热点 key 加随机前缀打散写入，聚合后再去掉前缀做二次聚合
- 作业调优：用 Spark UI 的 stage 页确认哪一步的 Shuffle Read/Write 最大，再决定动哪个参数
- 落库前重分区：写出文件数由 shuffle 分区数决定，按目标文件大小反推分区数
常见现象与动作
| 现象 | 含义 | 处理动作 |
|---|---|---|
| Shuffle Read 远大于计划数据量 | 分区不合理或过滤未下推到读取阶段 | 前置过滤、调整分区键 |
| 个别 task 耗时远高于均值 | 数据倾斜，单分区数据量过大 | 加盐打散或更换分区键 |
| 产生大量极小的 shuffle 分区 | 分区数固定得过大，调度开销占比高 | 开启 AQE 合并或下调分区数 |
| Executor 频繁 GC 甚至 OOM | 单分区数据量超过可用内存 | 提高分区数或加大 executor 内存 |
| 任务卡在最后几个 task | 少数热点分区拖尾 | 单独处理热点 key 或二次聚合 |
小表 join 大表优先广播（在 map 端完成连接），可彻底避免大 shuffle；但要确认小表真正放得进内存，广播阈值调高后一旦估错会直接 OOM。
分区数不是越多越好：分区过多会带来大量小任务与调度开销，分区过少则单任务数据量大、容易内存溢出。经验做法是让每个 shuffle 分区落在几十到两百 MB 的量级，再用自适应机制兜底。