Spark 3.0的AQE开了，还需要手动处理倾斜吗？

数据倾斜手册第 5 题约 10 分钟读完

本文目录

先说结论
AQE的三大核心能力
AQE Skew Join的工作原理
AQE处理不了哪些场景？
AQE和手动优化的配合策略
从Spark UI验证AQE是否在起作用
面试高频追问
实战踩坑
总结

先说结论

AQE能自动处理的：Shuffle之后的Join倾斜（Skew Join Optimization）和分区合并（Coalesce Partitions）。检测到某个Shuffle分区数据量远超中位数时，自动拆分大分区、复制对应侧的数据并行处理。

AQE处理不了的：GROUP BY聚合倾斜（尤其是COUNT(DISTINCT)）、Flink倾斜、数据源读取阶段的倾斜（如HDFS文件大小不均）、以及倾斜程度未达到AQE阈值的轻度倾斜。

建议：AQE是必开的兜底，但不能替代手动优化。两者结合：手动处理已知的严重倾斜，AQE兜底处理未知的轻度倾斜。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

AQE（Adaptive Query Execution，自适应查询执行）是Spark 3.0引入的重大特性，其中一个重要能力就是自动处理数据倾斜。很多人听到这里会觉得：开启AQE就完事了，数据倾斜问题自动解决了。但实际上，AQE能处理的场景有限，而且有自己的判断逻辑和阈值，理解这些才能真正用好它。这道题是区分「听说过AQE」和「真正用过AQE」的分水岭。

AQE的三大核心能力

AQE是Spark的运行时自适应机制，它在Stage执行完成后，根据实际的数据统计重新规划后续Stage的执行计划。三个核心能力：

能力	作用	解决的问题	默认是否开启
动态合并分区 Coalesce Partitions	把小分区自动合并，减少Task数量	过多Task的调度开销	AQE开启后自动开启
动态切换Join策略 Join Strategy Switch	运行时把Shuffle Join切换为Broadcast Join	Build表实际数据比预估小时的低效	AQE开启后自动开启
自动处理倾斜 Skew Join Optimization	检测并拆分倾斜的Shuffle分区	JOIN时的数据倾斜	需额外开启

# 开启AQE全部能力（放在Spark配置或SQL里）
spark.sql.adaptive.enabled = true                    # AQE总开关
spark.sql.adaptive.coalescePartitions.enabled = true # 动态合并分区
spark.sql.adaptive.skewJoin.enabled = true           # 倾斜Join处理
 
# 控制倾斜判断的阈值（重要！）
spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5   # 中位数的5倍才算倾斜
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256MB  # 且大于256MB
 
# 两个条件同时满足才触发AQE倾斜处理
# 调小这两个值，AQE会更积极地处理倾斜（但也可能处理不必要的情况）

AQE Skew Join的工作原理

理解原理，才能知道什么时候AQE有效、什么时候无效。

-- 假设这个JOIN存在倾斜
SELECT o.*, u.user_name FROM orders o JOIN users u ON o.user_id = u.user_id;
 
-- Spark执行过程（开启AQE）：
-- 1. Stage 1：对orders和users分别做Shuffle，分成N个分区
-- 2. AQE检测到orders的第5号分区数据量=1GB，其他分区平均50MB
--    1GB > 50MB*5=250MB ✓  且 1GB > 256MB ✓  → 判定为倾斜分区
-- 3. AQE把orders的第5号分区拆成4个子分区（250MB each）
-- 4. 相应地，把users的第5号分区复制4份
-- 5. 产生4个Task并行处理这1GB数据，每个Task只处理250MB
-- 最终结果正确（4个子Task结果合并），且不再有单个超大Task

这就是AQE的本质：它做了和「倾斜key单独处理」方案类似的事情，但是自动完成的，不需要你修改SQL。代价是：它需要先完成Shuffle（耗时），才能检测到倾斜并处理；而手动优化可以在Shuffle之前就减少数据量。

AQE处理不了哪些场景？

AQE不是万能的。以下场景它处理不了，必须手动优化：

-- 场景一：GROUP BY倾斜（AQE无效）
SELECT user_id, COUNT(DISTINCT event_type) AS type_cnt
FROM user_events
GROUP BY user_id;
-- AQE的Skew Join只针对JOIN操作，GROUP BY聚合的倾斜不在其管辖范围
-- 必须手动改写：子查询拆分COUNT(DISTINCT)
 
-- 场景二：倾斜程度未达阈值
-- 假设最大分区200MB，中位数50MB，factor=4（<5，未达阈值）
-- AQE不会触发，但这个倾斜仍然可能让作业慢3-4倍
-- 解决：降低 skewedPartitionFactor 阈值，或手动处理
 
-- 场景三：Broadcast Join本身是最优但AQE没有识别到
-- 如果统计信息不准确（比如建表时没有分析统计），
-- AQE的Join Strategy Switch也可能错误判断
-- 解决：手动用BROADCAST hint或提前ANALYZE TABLE

另外，AQE对流处理（Flink/Spark Streaming）无效——AQE是针对Spark批处理SQL的运行时优化。Flink的倾斜需要用两阶段聚合（局部+全局）或者给热点key做特殊路由来处理。

AQE和手动优化的配合策略

最好的实践不是「用了AQE就不用手动优化」，而是两者分工：

# 推荐的分工方式：
 
# 1. 线上作业必须开启AQE（兜底）
spark.sql.adaptive.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
 
# 2. 已知的严重倾斜，手动处理（AQE处理时机太晚，开销大）
#    - NULL值过多：手动过滤NULL
#    - COUNT(DISTINCT)：手动改写子查询
#    - 固定的热点key（如测试账号）：单独处理
 
# 3. 动态的、未知的轻度倾斜，交给AQE处理
#    适当调低倾斜阈值，让AQE更积极
spark.sql.adaptive.skewJoin.skewedPartitionFactor = 3  # 从5降到3
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 128MB  # 从256MB降到128MB
 
# 4. 定期用Spark UI观察AQE的效果
#    看DAG里是否有被标记为'Skewed'的分区
#    观察AQE拆分了多少个子Task

从Spark UI验证AQE是否在起作用

开了AQE怎么知道它在处理倾斜？去Spark UI看。

# 验证AQE倾斜处理的步骤：
 
# 1. 打开Spark UI → SQL标签页 → 点进具体查询
# 2. 查看DAG（查询执行图）
#    如果AQE处理了倾斜，会看到节点旁边有'(skewed)'标记
#    以及被拆分的额外Task（比如原来1个Task，AQE拆成4个）
 
# 3. 看Stage详情里的Task列表
#    正常：所有Task时间分布均匀
#    AQE处理后的倾斜Stage：Task数量增多，时间分布均匀（倾斜被消除）
 
# 4. 检查AQE的统计信息
# 在日志里搜索：
grep 'Skew join' spark-driver.log
# 或者：
grep 'Detected skewed partition' spark-driver.log
# 如果有这些日志，说明AQE在工作

面试高频追问

追问一：AQE的动态合并分区怎么用？和数据倾斜有关系吗？

AQE动态合并分区主要解决的是'小文件过多'问题，和数据倾斜是两个方向的问题。Shuffle之后可能产生大量小分区（比如设置了1000个分区，但数据量小，每个分区只有1KB），产生1000个Task的调度开销。AQE会把这些小分区自动合并（Coalesce），减少Task数量。而数据倾斜是某个分区特别大，AQE会拆分。一个是合并小的，一个是拆分大的，方向相反，但都是让分区大小更均匀，减少最大Task时间。

追问二：AQE开启后，shuffle.partitions还需要手动设置吗？

AQE开启动态合并分区后，spark.sql.shuffle.partitions的初始值仍然重要——它决定Shuffle初始分成多少份。AQE会在Shuffle完成后动态合并小分区，但不会把1个分区拆成多个（那是Skew Join干的事）。建议：shuffle.partitions设一个偏大的值（比如2000），AQE会自动合并小分区，同时Skew Join会拆分大分区，综合效果是各分区大小趋于均匀。如果shuffle.partitions设太小（比如默认200），数据量大时每个分区都很大，AQE没有合并空间，倾斜时拆分效果也有限。

追问三：AQE的实现代价是什么？会不会拖慢不倾斜的作业？

AQE有额外开销：1）每个Stage完成后需要收集统计信息（各分区大小），有短暂的延迟；2）重新规划执行计划有计算开销；3）倾斜处理时复制对应侧数据，有网络和内存开销。对于没有倾斜、数据分布均匀的作业，AQE的额外开销通常在1-3%以内，可以忽略。但对于Stage非常多（100+）的复杂作业，每个Stage都要收集统计，累积开销可能更明显。实际经验：绝大多数情况AQE的收益远大于开销，默认开启即可。

实战踩坑

我在某出海硬件品牌做Spark调优时，有一批作业从EMR迁到Databricks，上来就把AQE全开了（Databricks 10.x默认开启AQE）。迁完后有一个设备激活报表作业，理论上AQE应该处理掉倾斜，但作业还是跑了35分钟。

# 去Databricks的Spark UI查看，发现AQE没有触发倾斜处理
# 原因：倾斜分区 600MB，中位数 150MB
# 600MB / 150MB = 4倍，而默认 skewedPartitionFactor = 5
# 4 < 5，不满足条件，AQE没有介入
 
# 解决：降低阈值，同时分析为什么有这个倾斜
 
# 方法一：降低AQE阈值（让AQE更积极）
SET spark.sql.adaptive.skewJoin.skewedPartitionFactor = 3;
# 600MB / 150MB = 4倍 > 3倍 ✓  → AQE触发，作业从35分钟降到18分钟
 
# 方法二：同时分析倾斜根因
# 查到是某个device_type（工厂测试设备）数据量大
# 在SELECT前加过滤：WHERE device_type != 'FACTORY_TEST'
# 彻底消除倾斜后：作业从35分钟降到8分钟
 
# 教训：AQE降阈值是应急手段，根治倾斜才是最优解

复盘：AQE不是开了就完事，要理解它的判断逻辑（factor × 中位数 + 阈值大小），才能知道什么时候它会触发、什么时候不会。遇到AQE没有触发的倾斜，要么降低阈值，要么手动处理倾斜根因——两个都做效果最好。

总结

AQE是必开的兜底，能自动处理JOIN时的动态倾斜，还能合并小分区、动态切换Join策略。但它有局限：只针对JOIN不针对GROUP BY，有阈值门槛，流处理不适用。最佳实践是：AQE开起来做兜底，已知的严重倾斜（NULL倾斜、COUNT(DISTINCT)、固定热点key）手动处理。理解AQE的工作原理和阈值逻辑，是区分「听说过AQE」和「真正用过AQE」的关键。

下一篇聊Flink里的数据倾斜——流处理里的倾斜和批处理有什么不同？两阶段聚合在Flink里怎么写？

你们生产环境AQE开了之后，有没有遇到「AQE该触发但没触发」的情况？是怎么排查的？评论区分享～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

数据倾斜

面试通关手册

第6-10篇：Flink倾斜 · MapJoin · Hive倾斜 · 分桶预聚合 · 监控预警

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长

面试杀手题🔥🔥 | 进阶篇 | 每篇附实战案例

刷题进度保存在当前浏览器