Spark SQL 自适应执行AQE 开启前后查询计划差了多少大家好我是朱大喜。上周我们聊了平台架构这周进到执行引擎层面。你有没有遇到过明明数据量不大SQL 却跑了四十分钟的困惑问题大概率出在查询计划上。一、先看一个翻车现场五张表 JOIN 的灾难假设我们要做用户画像分析需要关联用户基础表、行为日志、订单表、商品表和活动表SELECT u.user_id, COUNT(DISTINCT o.order_id) as order_cnt, AVG(b.session_duration) as avg_session FROM users u JOIN behavior_log b ON u.user_id b.user_id JOIN orders o ON u.user_id o.user_id JOIN products p ON o.product_id p.product_id LEFT JOIN campaigns c ON u.last_campaign_id c.campaign_id WHERE b.dt 20260701 GROUP BY u.user_id;在AQE 关闭的情况下Spark 在 SQL 解析阶段就固定了执行计划。它根据表的统计信息预估每个分区的数据量决定用 SortMergeJoin 还是 BroadcastHashJoin。但问题是——统计信息经常不准graph LR subgraph AQE关闭: 固定执行计划 A1[SQL解析] -- A2[基于统计信息br/生成物理计划] A2 -- A3[join策略固定br/分区数固定br/无法中途调整] end subgraph AQE开启: 动态调整 B1[SQL解析] -- B2[生成初始计划br/预留调整空间] B2 -- B3[运行时收集br/实际数据量] B3 -- B4{数据量超过br/broadcast阈值?} B4 --|否| B5[临时切换br/BroadcastJoin] B4 --|是| B6[保持br/SortMergeJoin] end style A3 fill:#faa,stroke:#333 style B5 fill:#afa,stroke:#333 style B6 fill:#afa,stroke:#333类比一下AQE 关闭就像你出发前看了一眼高德地图然后全程不看导航闭眼开AQE 开启就像实时导航前面堵车了会帮你重新规划路线。二、AQE 的三大核心特性逐个拆特性一动态合并小分区Coalesce Shuffle Partitions这是 AQE 最朴实但也最实用的能力。Spark 默认 shuffle 分区数是 200如果你的数据只有 50MB分 200 个区意味着每个分区才 250KB——调度开销比计算开销还大from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(AQE_Demo) \ .config(spark.sql.adaptive.enabled, true) \ # 初始分区设大一点让 AQE 自己去合并 .config(spark.sql.adaptive.coalescePartitions.enabled, true) \ # 初始 shuffle 分区数设大是故意的给 AQE 足够空间去优化 .config(spark.sql.shuffle.partitions, 500) \ # 合并后单个分区的最小大小低于这个值的相邻分区会合并 # 设为 8MB 是因为 HDFS 一次读取的最佳块大小在这个量级 .config(spark.sql.adaptive.advisoryPartitionSizeInBytes, 8MB) \ .getOrCreate() # 场景对用户行为日志做聚合数据倾斜严重 # 大 V 用户的数据量是小用户的 1000 倍 df spark.sql( SELECT user_id, COUNT(*) as cnt FROM behavior_log WHERE dt 20260701 GROUP BY user_id ) # AQE 的实际效果 # 1. 先按 500 个分区 shuffle够细 # 2. 运行中发现大部分分区数据量很小自动合并到 ~50 个分区 # 3. 减少 task 调度开销总耗时从 180s 降到 45s df.explain(formatted)为什么spark.sql.shuffle.partitions初始要设大500/1000而不是设小比如 50这是 AQE 的底层设计决定的——它只能合并、不能拆分。AQE 的小分区合并逻辑叫 Coalesce它是把相邻的多个小分区聚合成一个大分区但它没有把一个大分区切分成多个的反向能力那是倾斜 Join 优化的专属技能。所以策略就是先散后聚初始 500 个分区足够细按 key 哈希分布比较均匀AQE 运行阶段合并到合适大小比如 128MB 一个分区最终可能 30-50 个分区。如果你初始只设 50 个分区AQE 也救不了——50 个分区里如果有 1 个大 partition 撑死200MB其他 49 个空着AQE 没法把 200MB 的大分区切开分给其他 task。一句话初始设大是留给 AQE 的优化空间设小就是堵死优化的路。特性二动态切换 Join 策略这是 AQE 最值钱的能力。考虑这个场景-- 订单表 500 万行缓存表 10 万行 -- 统计信息显示缓存表有 50 万行过期了 -- 所以 Spark 用了 SortMergeJoin SELECT o.*, c.cache_hit FROM orders o JOIN cache_log c ON o.user_id c.user_id;统计信息说缓存表 50 万行超过了spark.sql.autoBroadcastJoinThreshold默认 10MB所以 Spark 用了 SortMergeJoin——需要 shuffle 两张表磁盘读写开销巨大。但实际上缓存表只有 10 万行AQE 开启后Spark 会在运行时读取缓存表实际大小发现只有 3MB直接将 SortMergeJoin降级为 BroadcastHashJoin——把缓存表广播到每个 executor省掉了订单表的 shuffle。# 开启动态 Join 策略切换 spark.conf.set(spark.sql.adaptive.autoBroadcastJoinThreshold, 20MB) spark.conf.set(spark.sql.adaptive.localShuffleReader.enabled, true) # 更激进的方案对特别倾斜的 key 单独处理 spark.conf.set(spark.sql.adaptive.skewJoin.enabled, true) # 当某个分区的数据量超过中位数的 5 倍时自动拆分 # 为什么是 5 倍经验值低于 5 倍的开销不值得拆分 spark.conf.set(spark.sql.adaptive.skewJoin.skewedPartitionFactor, 5)特性三倾斜 Join 优化数据倾斜是 Spark 性能的头号杀手。比如 user_id NULL 的记录有 3000 万条全进了一个分区。graph TD subgraph 倾斜前 S1[Task1: user_001br/100万行] -- S2[运行时间: 30s] S3[Task2: NULL值br/3000万行] -- S4[运行时间: 45minbr/其他task干等着] S5[Task3: user_003br/100万行] -- S6[运行时间: 30s] end subgraph AQE倾斜优化后 T1[Task1: user_001br/100万行] -- T2[30s] T3A[Task2-A: NULL值br/1000万行] -- T4A[15min] T3B[Task2-B: NULL值br/1000万行] -- T4B[15min] T3C[Task2-C: NULL值br/1000万行] -- T4C[15min] T5[Task3: user_003br/100万行] -- T6[30s] end style S4 fill:#faa,stroke:#333 style T4A fill:#afa,stroke:#333 style T4B fill:#afa,stroke:#333 style T4C fill:#afa,stroke:#333AQE 的倾斜处理逻辑是检测到某个分区数据量超过分区中位数 × skewedPartitionFactor时将其拆分为多个子分区分别与另一张表做 Join最后合并结果。为什么 AQE 用中位数而非均值来检测倾斜这是一道分布式系统设计里的统计直觉题。在数据倾斜的场景下1 个超大分区的存在会把均值拉高——比如 99 个分区平均 10MB1 个分区 5GB均值瞬间被拉到约 60MB按均值×5300MB 来检测5GB 的分区确实能被标出来。但问题在于方向相反的倾斜如果你的表有 30% 的分区数据偏小1KB均值会被拉低导致正常大小的分区比如 50MB被误判为倾斜——这就是均值对离群值的双向敏感性。而中位数只关心中间那个分区多大不受 1 个 5GB 分区或 30% 的 1KB 分区影响它能给你一个真实的正常分区该多大的基准。skewedPartitionFactor5的含义是只有偏离这个基准 5 倍以上的分区才配被拆不冤枉正常分区不漏掉真正倾斜的 key。三、生产环境 AQE 配置清单下面是一份经过我们团队验证的 AQE 配置模板可以直接拿去用# AQE 生产环境推荐配置 # 【总开关】不开白不开Spark 3.0 默认开启 spark.conf.set(spark.sql.adaptive.enabled, true) # 【小分区合并】初始设大让 AQE 自己缩 # 为什么初始要设大SQL 解析阶段无法预知上游数据量 # 设大了 AQE 能合并设小了没法拆拆了也没意义 spark.conf.set(spark.sql.shuffle.partitions, 500) spark.conf.set(spark.sql.adaptive.coalescePartitions.enabled, true) # 目标分区大小和 HDFS block 对齐128MB减少小文件 spark.conf.set(spark.sql.adaptive.advisoryPartitionSizeInBytes, 128MB) # 【动态 Join】这三个要一起开 spark.conf.set(spark.sql.adaptive.localShuffleReader.enabled, true) # Broadcast 阈值看你的 executor 内存一般设为 executor 内存的 1/4 spark.conf.set(spark.sql.adaptive.autoBroadcastJoinThreshold, 50MB) # 倾斜 Join数据倾斜是常态就得开 spark.conf.set(spark.sql.adaptive.skewJoin.enabled, true) # 【勿动项】以下参数一般不建议改 # coalescePartitions.minPartitionNum: 默认 1改了可能导致并行度不足 # skewJoin.skewedPartitionThresholdInBytes: 默认 256MB够用了四、AQE 不是万能药它的盲区AQE 再智能也有搞不定的情况盲区一统计信息本身就是空的如果你的表从来没有执行过ANALYZE TABLEAQE 的初始决策会非常保守——它需要先跑一个 stage 才能收集到真实数据量。解决办法很简单-- 定期更新统计信息让 AQE 有个准确的起点 ANALYZE TABLE behavior_log COMPUTE STATISTICS; ANALYZE TABLE behavior_log COMPUTE STATISTICS FOR COLUMNS user_id, dt;盲区二UDF 里的黑盒AQE 只能优化 Spark 内置算子的物理计划如果你的 SQL 里用了大量自定义 UDFSELECT user_id, my_custom_decode(payload) as decoded FROM events;这个my_custom_decode的计算开销 AQE 完全感知不到自然无法优化。盲区三文件层面的问题AQE 运行在 RDD/shuffle 层面如果你的数据源有 10 万个 1KB 的小文件AQE 帮不了你——那是存储层要解决的问题该用 Delta Lake 或定期 compaction 了。 踩坑提醒spark.sql.adaptive.coalescePartitions.minPartitionNum别乱改— 这个参数默认是 1意思是 AQE 可以把所有分区合并成 1 个。有人觉得合并越多越快把它设成 1 个分区——结果下游操作全变单 task并行度归零。保持默认值就好让 AQE 按数据量自主决定。倾斜 Join 优化只对 SortMergeJoin 生效— 如果你的 Join 已经被优化成了 BroadcastHashJoinskewJoin.enabled设多少都没用。所以倾斜 Join 的最佳实践是先检查你的数据确实走的是 SortMergeJoin再开这个参数否则就是空自转。AQE 的首个 Stage 必须跑完才能做优化决策— 这意味着 AQE 的收益在长链路 SQL3 个 Stage 以上中最明显。如果你的 SQL 只有 1 个 Stage单表聚合、无 JoinAQE 完全没机会介入——不要在短 SQL 上纠结 AQE 配置。五、总结AQE 的核心价值一句话让查询引擎有能力在运行时反悔编译时做的错误决策。三个关键能力动态合并分区解决分区太多调度比计算久的问题动态切换 Join 策略解决统计信息过期导致选错 Join 方式的问题倾斜 Join 优化解决一个慢 task 拖垮整个 stage的问题。配置建议也简单Spark 3.2 全开 AQE默认就是开的关键是把spark.sql.shuffle.partitions调大500~1000让 AQE 有优化空间。唯一需要额外关心的是定期ANALYZE TABLE让统计信息别太离谱。下篇我们聊聊 ClickHouse 的高可用看看副本同步延迟怎么排查和止损敬请期待