Spark SQL 数据倾斜性能调优实战

数据倾斜是 Spark SQL 大查询性能瓶颈的头号原因。一个包含 10 亿条记录的 join 查询，如果 key 分布不均，最慢的 task 可能比最快的慢 100 倍，而整体查询时间取决于最慢的 task。调优的核心思路是"少读、少传、少倾斜"三层递进，在 Spark 3.x 的 AQE（Adaptive Query Execution）框架下有更系统的解法。

一、少读：减少数据扫描量

倾斜调优的第一步不是解决倾斜本身，而是确保没有读入不必要的数据。很多看似"倾斜"的查询，实际上是大量无关数据被拉入了 Shuffle 阶段。

分区裁剪： 确认 WHERE 条件中的分区列被正确下推。在 Spark 3.x 中，默认开启 spark.sql.sources.partitionOverwriteMode=dynamic，但如果你的查询使用了 INSERT OVERWRITE，需要确认分区裁剪是否生效。用 EXPLAIN 查看执行计划，确认 Scan 节点上是否显示了 PartitionFilters。如果分区列是字符串类型而过滤条件是整数类型，类型不匹配会导致裁剪失效，所有分区的数据都会被读入。

谓词下推： Spark 3.x 会将过滤条件下推到数据源（如 Parquet、ORC）。但如果过滤条件涉及类型转换（比如字符串列与整数比较），谓词下推可能失效。确保过滤条件的类型与列类型一致。另外，如果过滤条件使用了 UDF 或复杂表达式（如 DATE_FORMAT(date_col, 'yyyy-MM') = '2026-08'），谓词下推也会失效。将日期过滤改为范围条件（如 date_col >= '2026-08-01' AND date_col < '2026-09-01'）可以恢复下推。

列裁剪： 避免 SELECT *，只读取需要的列。对于 Parquet 和 ORC 列式存储格式，列裁剪可以直接跳过不需要的列的数据块，减少 IO 和内存占用。在 Join 查询中，确保 SELECT 列表中只包含需要的字段，避免将大字段（如 JSON 字符串、数组类型）带入 Shuffle 阶段。

二、少传：减少 Shuffle 数据量

Shuffle 是数据倾斜产生性能影响的主要阶段。减少 Shuffle 数据量的手段包括：

Broadcast Join： 当 join 的一侧表较小时（通常小于 100 MB），将其广播到所有 executor 节点，避免 Shuffle。Spark 3.x 默认会自动判断是否使用 broadcast join，阈值由 spark.sql.autoBroadcastJoinThreshold 控制（默认 10 MB）。如果 join 的小表超过默认阈值但仍然可接受，可以手动调大该值：

SET spark.sql.autoBroadcastJoinThreshold=100m;

注意：broadcast join 会将整张表复制到每个 executor 的内存中，如果 executor 内存不足会导致 OOM。调大阈值前先确认 executor 的可用内存。如果你有多个 join 操作，broadcast join 只能应用于其中一个表——Spark 会自动选择最优的 join 顺序，但你可以通过 /*+ BROADCAST(table_name) */ hint 强制指定广播哪张表。

聚合下推： 在 group by 之前先做一次局部聚合（partial aggregation），减少 Shuffle 传输的数据量。Spark 3.x 的 AQE 会自动在某些场景下插入 partial aggregation，但手动编写 GROUP BY 时可以在子查询中先按相同 key 做一次 COUNT 或 SUM，再在外层做最终聚合。这种"两阶段聚合"策略对 count、sum 等可合并聚合函数效果显著，但对 avg 等不可直接合并的函数需要先 sum 再 count 来计算。

三、少倾斜：针对 Skew Key 的精确处理

当数据分布本身不均匀时，需要针对倾斜 key 做特殊处理：

Spark 3.x AQE Skew Join： AQE 提供了自动倾斜处理能力，核心参数如下：

• spark.sql.adaptive.enabled=true：开启 AQE
• spark.sql.adaptive.skewJoin.enabled=true：开启倾斜 join 优化
• spark.sql.adaptive.skewJoin.skewedPartitionFactor=5：如果某个分区的大小超过中位数的 5 倍，判定为倾斜分区
• spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256m：倾斜分区的最小大小阈值

开启后，AQE 会在运行时检测到倾斜分区，并将倾斜的大分区拆分为多个小分区，分配给不同的 task 并行处理。这个过程是透明的，不需要修改 SQL。AQE 在 Shuffle 完成后才做倾斜检测，因此它能精确地基于实际数据量判断倾斜程度，而不是依赖静态的统计信息。

盐值打散（Salting）： 对于 AQE 无法自动处理的极端倾斜（比如某个 key 的数据量占总量的 50%），可以在 join 前对倾斜 key 添加随机后缀，将其分散到多个分区中。具体做法是对倾斜 key 拼接 0 到 N 的随机后缀，同时对小表的对应 key 也做膨胀处理（行数乘以 N），然后去掉后缀做 join。盐值的 N 值通常设为 10 到 50，根据倾斜程度和数据量调整。需要注意的是，盐值打散会增加 Shuffle 数据量（小表被膨胀 N 倍），因此只适用于倾斜严重到 AQE 无法处理的场景。

动态分区合并： AQE 的 spark.sql.adaptive.coalescePartitions.enabled=true 会在 Shuffle 后自动合并过小的分区，减少小文件和过多 task 导致的调度开销。这对于倾斜处理后的分区尤其有用——拆分倾斜分区后可能产生大量小分区，自动合并可以将它们重新聚合为合理大小。

四、参数调优参考

以下参数按"先基础后进阶"排列，建议逐个启用并观察效果。

基础参数： spark.sql.shuffle.partitions（默认 200）控制 Shuffle 后的分区数。对于大表 join，建议将其设为数据量（MB）除以 64 到 128。比如 10 GB 数据建议设为 80 到 160。过大的分区数会导致小分区和调度开销，过小会导致单分区数据量过大而倾斜。

进阶参数： 开启 AQE 后设置 spark.sql.adaptive.advisoryPartitionSizeInBytes=64m 控制重组分区的建议大小，spark.sql.adaptive.coalescePartitions.minPartitionNum=1 允许 AQE 将分区合并到最少 1 个。

极端参数： 开启 skewJoin 后，如果仍然有单个 task 超时，可以临时将 skewedPartitionFactor 从 5 调到 3，让更多分区被判定为倾斜。但要注意，更激进的阈值会增加 AQE 的处理开销，在数据量极大时可能反而拖慢查询。

五、监控与验证

调优后需要验证效果。通过 Spark UI 的 SQL 标签页查看执行计划，重点关注 Shuffle Read 和 Shuffle Write 列：如果所有 task 的 Shuffle Read 接近，说明倾斜已消除。用 EXPLAIN FORMATTED 查看完整的物理计划，确认 broadcast join、AQE 合并等优化是否生效。如果发现某个 task 的 Shuffle Read 仍然远大于其他 task，可以通过 Spark UI 的 Stage 详情查看该 task 处理的具体分区，定位是哪个 key 导致的倾斜。此外，建议在生产环境的调度任务中加入 Spark UI 指标的自动化采集，记录每次查询的 task 分布标准差，形成调优前后的量化对比数据。

EasyMR 集成 Hive、Spark、Flink 等批流计算引擎，基于 Hadoop 自研增强，统一承载批计算与 OLAP 场景，可以直接使用 Spark 3.x 的 AQE 特性进行数据倾斜调优，无需额外的平台适配工作。

调优路径总结

Spark SQL 数据倾斜调优不是单一参数的调整，而是"少读减少数据量、少传减少 Shuffle、少倾斜精确拆分"三层递进的系统工程。先确保分区裁剪和列裁剪正常工作，再通过 broadcast join 和聚合下推减少 Shuffle 数据量，最后用 AQE 自动倾斜处理或盐值打散解决顽固的 key 分布不均问题。每次调整后通过 Spark UI 验证效果，形成闭环。

如果你的大查询频繁出现单个 task 执行时间远超其他 task 的情况，建议先用 EXPLAIN 确认执行计划中是否存在 Shuffle，然后开启 AQE 的 skewJoin 参数，最后根据 Spark UI 中的 task 分布决定是否需要盐值打散。实际调优中往往需要多轮迭代——先解决最大的倾斜来源，观察效果后再处理次要的性能瓶颈，逐步将查询性能推向最优。