企业官网建设流程全解析

提效7倍，Apache Spark自适应查询优化的深度实践与改进

我大概两年前开始接手集群上一批日调度任务的性能治理工作，当时最头疼的就是那些动不动跑四十分钟到一个小时的离线任务。查过Spark UI、调过并行度、改过存储格式，效果有，但总觉得是在“拆东墙补西墙”。后来系统性地把Apache Spark自适应查询优化（AQE，Adaptive Query Execution）吃透，并且在几类典型业务场景里陆续落地之后，一批任务的运行时间从四十五分钟直接压到了六分钟左右。这个效果不是靠某一条SQL技巧或者某个单一参数堆出来的，而是把AQE的机制跟业务数据特征真正匹配起来之后才出现的。这篇文章就把我在这个过程中的理解、踩过的坑、调参细节和最终的执行方案完整写出来，希望能帮到正在被Spark性能问题折磨的人。

1. 项目概述：为什么我们盯上了自适应查询优化

1.1 大数据任务性能瓶颈的共性痛点

先说说当时的背景。我们集群上跑着大量基于Apache Spark的离线数据加工任务，数据量级在TB级别，每天凌晨开始调度，高峰期有几百个任务同时跑。最典型的问题集中在以下几个方面：

第一，数据倾斜。看似简单的join，因为某个业务key分布不均匀，最终导致少数几个task要处理几亿条记录，而其他task早就跑完在等。整个stage的耗时取决于那个最慢的task，这就是所谓的“长尾效应”。早期我们统计过，集群里超过一半的任务都存在不同程度的倾斜问题，只是有的严重有的轻微。

第二，执行计划不准确。Spark SQL优化器在生成执行计划时，依赖的是表级别的统计信息，比如行数、字节数、列分布。但真实生产环境里，很多表根本来不及做ANALYZE TABLE，或者数据每天都在变，昨天分析的统计信息今天就已经失真。统计信息失真，优化器就可能选错join策略、算错shuffle分区数，生成的执行计划跟实际数据情况完全对不上。

第三，shuffle分区数的静态设定。以前很多人习惯用spark.sql.shuffle.partitions=200这类全局配置。这个值设小了，单个task处理的数据量过大；设大了，几十亿条数据被切成几千个甚至上万个碎片，下游又有几千个空任务白白启动，调度开销和序列化开销都很大。实际数据分布是每天都在变化的，一个静态值根本不可能在任意时刻都合适。

这些问题单独挑任何一个出来都有对应的“土办法”去救，比如给某几个key加盐、手动重分区、调整并行度。但问题在于，这些手段都属于“面向具体的某一个任务去特判”，换个任务、换一天的数据，同样的特殊处理就失效了。所以我们需要一个能从机制层面去自适应匹配数据变化的方案，这就是我决定深入研究和落地AQE的根本原因。

1.2 自适应查询优化带来的核心转变

Apache Spark自适应查询优化并不是一个单一的功能，而是一套在任务执行过程中动态优化执行计划的机制。它最核心的思想是：不再完全依赖执行前的静态统计信息，而是把“优化”这件事延后到shuffle阶段，在拿到真实数据统计之后再做出决策。

Spark在3.0版本开始正式把AQE作为默认开启的特性引入。它的能力覆盖三个典型场景：动态合并shuffle分区、动态调整join策略、动态优化数据倾斜join。这三板斧正好对应了我们刚才提到的三大痛点。

这里我想用一个比较形象的比喻。传统的Spark优化器像是“盲人摸象”，在执行之前根据一张模糊的照片去判断大象的样子，结果经常判断错。而AQE像是“摸着石头过河”，走到河中间，摸到了真实的石头位置，再决定下一步怎么落脚。这个“边执行边修正”的思路，是Spark在执行模型上的一次比较重要的演进。

因此我在项目立项时，目标就很明确了：不是简单给每个任务开一个配置开关，而是要把AQE的三项能力用在最匹配的业务场景里，再配合必要的参数调整，真正把任务的执行流程从静态变成动态。

2. AQE核心原理拆解：三个关键机制

2.1 动态合并Shuffle分区：从“一刀切”到“按需分配”

先来看AQE最基础、也最容易被感知到的一项能力：动态合并shuffle分区。

在Spark 2.x时代，shuffle分区的数量在任务开始前就固定了。比如你设置了spark.sql.shuffle.partitions=200，那么不管你的数据实际有多大，哪怕只是一个几百MB的小join，也会强行切成200个分区去跑。分区数量过多的问题我之前提过：每个分区对应一个下游task，每个task都有启动、调度、序列化、反序列化的开销，这些开销在小数据量下被放得特别大。

AQE的做法是：在shuffle的map端任务写完数据之后，框架会先统计每个分区实际的数据大小，然后根据用户配置的目标分区大小（spark.sql.adaptive.advisoryPartitionSizeInBytes，默认64MB），去计算一个“比较合理的最终分区数”。如果某几个分区都很小，就把它们合并成几个更大的分区。这个合并过程在shuffle read阶段发生，下游stage的任务数会明显减少。

举个例子。一个join任务，map端写出的数据总量约1.5GB，如果静态设置800个分区，每个分区平均只有不到2MB的数据，这些数据会被分散到800个task里，每个task要处理的任务碎片非常多，调度压力大但每个task的实际计算量非常小。AQE开启后，会按照目标分区大小64MB来算，大约只需要24个分区，下游task数直接降了一个数量级。

我在实际使用中的体会是，分区合并带来的收益并不是“减少计算量”，而是“减少任务调度和序列化开销”，同时还能减少shuffle read阶段的随机IO压力。尤其是在上游map输出特别碎片化、分区数设置得特别夸张的场景下，收益最明显。

2.2 动态切换Join策略：让优化器“看清”之后再决定

第二个机制是动态调整join策略。这里重点说shuffle join和broadcast join之间的切换。

正常情况下，Spark优化器在决定两个表怎么join时，会看spark.sql.autoBroadcastJoinThreshold这个参数（默认10MB）。如果优化器判断某张表小于这个阈值，就会把它做成broadcast变量广播到各个节点，用BroadcastHashJoin去执行，避免shuffle。但问题的关键是，优化器这个判断依赖的统计信息，不一定准。如果一张表的统计信息长期没更新，或者实际上已经膨胀到了几百MB，优化器依然可能拿旧信息做判断，导致选择了错误的策略。

AQE的调整方式是：join的shuffle map端执行完之后，框架能拿到shuffle写入数据实际大小，用这个真实值重新判断表的大小，再决定是否把SortMergeJoin切换成BroadcastHashJoin，或者反过来。注意这里有个细节，动态切换join策略不只是“把小表切换成广播”，还包括“当小表实际并不小时，取消广播计划，改用shuffle join”，防止广播一张超大表造成Driver内存爆掉。

举一个我遇到的实际场景。某张维度表，昨天全量更新后只有约8MB，优化器判断它小于阈值10MB，于是生成了broadcast join计划。结果当天数据源异常，产出的维度表膨胀到了300MB，但由于统计信息没有及时更新，任务依然按广播策略执行。最终导致每个executor都要加载一份300MB的广播变量，内存占用飙升，任务频繁GC甚至OOM。AQE在这种情况下会基于shuffle后真实的size判断，认为该表不再适合广播，于是动态切换回SortMergeJoin。这个“纠正”的能力，比单纯依赖静态阈值要稳健得多。

2.3 动态优化数据倾斜Join：把“长尾”任务拆掉

第三个机制也是对我们这批任务提升最大的机制：动态优化数据倾斜join。

这个机制解决的就是我在开头说的那个痛点：某个key的数据量特别大，导致某个分区拖垮整个stage。传统的解决办法是手动加盐（salting），但加盐需要业务理解——你至少得知道哪个key可能倾斜，还要针对不同的数据分布去设计不同的加盐方式。而AQE的skew join优化，是直接在shuffle阶段自动检测并拆分的。

它的原理可以这样理解：当shuffle map端写完数据之后，AQE会检查每个分区的大小，如果发现某个分区的大小超过了一个阈值（spark.sql.adaptive.skewJoin.skewedPartitionFactor乘以分区中位数，同时还要大于skewedPartitionThresholdInBytes的最小值），并且这个分区的大小是整个分区中位数的好几倍，那么框架就会把这个大分区拆成多个更小的子分区，让它们在多个task上并行执行。这样，原来由一个task处理的几亿条数据，就被拆成多个task各自处理其中一小块，处理时间自然大幅下降。

这里有个需要注意的点：AQE的倾斜join优化，并不是对所有的倾斜都能生效。它主要针对的是shuffle join中某个分区数据量过大，但该分区内的key仍然可以继续拆分的情况。如果某个key本身语义上不能拆，比如你要做聚合、去重、排序，或者倾斜key本身导致后续处理存在依赖，那单纯拆分区不一定能解决问题。这种情况下还是需要结合业务场景做更底层的处理。

3. 落地实践与关键参数配置

3.1 开启AQE的环境与版本要求

先明确一点：AQE是Spark 3.0起正式合入的特性，需要把Spark版本升级到3.0及以上才能使用。项目里如果还在用Spark 2.3、2.4这样的老版本，就得先做版本升级。我当时的实践环境是Spark 3.1.2，后来部分任务也跑过3.2和3.3，整体功能差异不大但参数细节略有区别。

开启方式非常简单，在提交任务时加上以下配置：

spark.sql.adaptive.enabled=true

如果是Spark 3.2及以上，AQE默认就是开启的。但默认开启不等于默认好用，因为默认参数面向的是“通用场景”，而真实业务数据的特征往往千差万别，必须做针对性的参数调整，才能真正吃到收益。

3.2 核心参数清单与调参经验

这里整理一份我实际用过、并且经过对比验证的核心参数清单，方便直接参考：

有几个参数值得多说几句。

parallelismFirst这个参数，默认是true，意思是动态合并分区时优先保证并行度，宁可分区多一些也不让分区太大。实际调优时，如果任务的总数据量比较稳定，资源也比较充足，我会倾向于把这个参数设为false，让分区大小更贴近advisoryPartitionSizeInBytes的目标值，减少空转task。反过来，如果任务的资源并行度非常关键，或者下游计算特别重，保留高并行度是更安全的选择。

advisoryPartitionSizeInBytes这个值不建议盲目调大。我在一个任务上调到256MB，结果分区倒是减少了，但每个task处理的shuffle数据过大，导致GC时间明显增加。后来回到128MB，效果反而更好。什么东西叫最优，要看你的executor内存和单个task的计算复杂度，建议用实际的A/B测试来判断，不要凭感觉。

skewJoin.skewedPartitionFactor和skewedPartitionThresholdInBytes是一对组合条件：只有当某个分区大小超过中位数乘以factor，并且自身大小还超过threshold时，才会被判定为倾斜分区。如果只调大factor，可能会导致部分倾斜分区逃过检测；如果只调大threshold，则阈值设置过高后，中等程度的倾斜也不会被优化。这两个参数需要配合数据分布特征来定。

3.3 一个真实任务的优化前后对比

拿我们集群上一个最典型的任务来举例。这是一张订单事实表和一张用户维度表的join加工任务，数据量大约2TB，原执行计划是SortMergeJoin，shuffle分区数静态配置为800，存在比较明显的用户维度倾斜。

优化前的情况是：整个任务耗时约45分钟，其中大部分时间都耗在join前的shuffle读取和最后的倾斜task上。从Spark UI上看，同一个stage里面，部分task运行时间不到两分钟，但有几个task跑了将近30分钟，典型的倾斜形态。

开启AQE后，我对这个任务做了如下调整：

SET spark.sql.adaptive.enabled=true; SET spark.sql.adaptive.coalescePartitions.parallelismFirst=false; SET spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB; SET spark.sql.adaptive.skewJoin.skewedPartitionFactor=5; SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB;

任务跑完后的效果非常直观：shuffle分区数从800动态降低到了60多个，倾斜的大分区被自动拆分到多个task并行处理，整个任务的执行时间从45分钟缩短到了大约6分钟。这个过程中我没有改任何SQL逻辑，也没有给某些key手动加盐，全部收益都来自AQE的动态优化。

另外观察Spark UI时，有一个比较明显的标志：开启AQE之后，shuffle read阶段会多出一些“Optimized”的标记，比如Optimized Local Shuffle Reader，同时stage的task数量会跟代码里设置的分区数不一致，而是动态调整后的数量。这些都可以用来确认AQE真的生效了。

4. 性能提升7倍的实现路径与改进

4.1 从数据特征出发的针对性优化

虽然AQE像一个“万能开关”，但真正的性能提升不是靠一键开启就能实现的，还得针对数据特征做细致的分析。这里我梳理了几个还比较有效的步骤。

第一步，先做数据画像。对一个性能较差的Spark任务，先看它的执行计划，重点看有几处shuffle、shuffle的数据量有多大、哪些stage耗时占比最高。再结合Spark UI查看task运行分布情况，判断是否存在倾斜。这一步非常关键，因为如果任务的瓶颈不在shuffle和join，而是在某个UDF计算过于复杂，那么AQE帮不了你太多。

第二步，根据瓶颈类型选择AQE的优化方向。如果瓶颈是shuffle分区数过多造成的调度开销大，重点是开启动态合并，并调整advisoryPartitionSizeInBytes。如果瓶颈是表的大小导致join策略选错，重点是动态调整join策略。如果瓶颈是数据倾斜造成的长尾，重点是开启动态skew join优化，并调好倾斜检测阈值。

第三步，做A/B测试。不要只对比“开没开AQE”，至少要做三组：关闭AQE的基线、开启AQE但全默认参数、开启AQE且做了针对性调参。我遇到过一种情况，开启AQE后任务反而变慢了，原因就是默认配置下动态合并分区合并得太狠，导致一个task要处理的数据量过大，整体性能不升反降。如果没有A/B测试，很容易得出“AQE没效果”的错误结论。

以其中一个任务为例，我做过的三组对比数据如下：

这组对比说明，AQE的默认参数确实有收益，但不一定是最优的，只有针对业务场景做了调整之后，才能真正把潜力释放出来。

4.2 常见问题排查实录

实践过程中当然不是一路顺畅，下面几个问题我基本都踩过，并且花了不少时间排查，分享出来帮大家少走弯路。

第一个问题是：开启AQE后任务反而变慢了。刚开始排查时也很困惑，后来发现是把parallelismFirst设为false之后，某个任务被合并成了少数几个大分区，每个分区的数据量超过了单task能高效处理的量级，导致执行时间反而上升。解决思路是不要只看单个参数，要把minPartitionNum、advisoryPartitionSizeInBytes和maxNumPostShufflePartitions三者放在一起综合考虑，确保合并后的分区数既不要太多，也不要太少。

第二个问题是：动态广播join导致Driver OOM。AQE动态切换join策略时，如果某张被切成shuffle的表实际上并不小，但因为nonEmptyPartitionRatioForBroadcastJoin这个参数设置不合理，导致框架错误判断为可以广播。我在一个任务里把这张参数调到了0.5，结果一张30GB的表被当成广播对象，Driver端内存直接被打爆。后来把这个参数降回0.2，同时把spark.sql.adaptive.localShuffleReader.enabled保持在true，问题就解决了。

第三个问题是：倾斜join优化没有生效。排查发现有两个原因：一是倾斜的阈值设置得太高，某个分区虽然相对别人已经很大，但还没有超过skewedPartitionThresholdInBytes，所以没有被识别；二是倾斜的分区并不是出现在shuffle join阶段，而是在groupBy聚合阶段，这种情况AQE的skew join机制是管不到的，需要手动干预。

第四个问题是：AQE的不稳定导致血缘变化。有些下游任务依赖Spark SQL的物理执行计划做血缘分析，AQE开启后，由于执行计划是动态的，血缘采集会变得不稳定。解决办法是把AQE生成的优化后执行计划采集回MetadataStore，或者调整血缘解析逻辑，兼容动态计划。

这些问题的共同点是：AQE不是“配置完就完事”，它需要结合业务场景去调优，并且要关注它带来的副作用，比如内存压力、血缘变化、资源竞争等。

4.3 排查与调优工具整理

调优过程中，有几个工具和技巧非常实用，整理在这里供大家参考。

第一，Spark UI的SQL Tab和Stage Tab是主要战场。在SQL Tab里可以看到每个节点的真实数据量，比如shuffle write和shuffle read的大小。如果发现shuffle read阶段存在“Optimized”标记，说明AQE的合并或本地读优化生效了。在Stage Tab里重点看task的运行时间分布，如果大多数task在一两分钟内完成，但有少数task要跑几十分钟，倾斜的判断基本就坐实了。

第二，Event Log是排查历史任务的好帮手。打开spark.eventLog.enabled=true，同时把spark.eventLog.dir配置到HDFS或对象存储上。这样每个任务的执行计划、每个stage的指标、每个task的运行时间和shuffle量都能完整记录下来，出问题时可以回溯。

第三，针对参数调优，建议用spark.sql.adaptive.*这类配置的前缀在Spark官网文档里把所有参数捞一遍，明确各个参数的默认值和相互影响关系。调参时每次只改一个参数，记录效果，不要多个一起改，否则出了问题很难定位。

第四，如果平时用了数据湖或Catalog服务，可以在表级别做ANALYZE TABLE，把统计信息维护好。虽然AQE能够动态纠偏，但精准的静态统计信息可以让优化器在初始阶段就生成更好的计划，减少AQE兜底的压力。

5. 经验总结与个人体会

5.1 容易被忽视的细节

在把AQE推广到更多任务之后，我总结了几个容易被忽视但特别重要的细节。

第一个细节是：AQE的倾斜优化对“groupBy”无效。虽然AQE能处理shuffle join的数据倾斜，但对于group by/distinct类的聚合倾斜，它是不介入的。因为这类聚合的结果不能简单拆分，分区拆了也无法并行去重。遇到这类倾斜，还是得靠两层聚合、加盐、或者改进算法来解决。所以不要以为开了AQE就万事大吉，倾斜问题依然只解决了一半。

第二个细节是：动态合并分区对“有状态”操作需要谨慎。比如流式任务或连续执行计划里的有状态算子，如果依赖固定分区数来维护状态，动态合并可能会导致key的分布变化，影响状态管理。在我自己的实践里，流式任务一般不动coalescePartitions.enabled，保持分区相对稳定。

第三个细节是：广播阈值和资源规格需要联动。AQE动态切换join策略时，广播变量需要占用driver端和executor端的内存。如果executor内存本身就比较紧张，把广播阈值调大很容易诱发OOM。建议在配置广播阈值之前，先评估executor内存大小和并行task数量，做到心中有数。

第四个细节是：AQE收益最大的场景，往往是资源充裕且数据特征多变的离线批任务。如果一个任务本身运行时间只有一两分钟，或者数据量非常小，AQE的动态优化机制带来的收益很有限，甚至可能因为额外统计开销造成轻微负优化。所以建议优先把AQE应用到耗时几十分钟以上的重型任务上。

5.2 后续扩展想法

AQE在Spark 3.x里已经比较成熟，但也有不少可以继续扩展和实践的方向。

我现在正在尝试的方向是，把AQE和动态资源分配（Dynamic Resource Allocation）结合起来。AQE决定任务需要的分区数，动态资源分配决定集群给任务提供多少个executor，两者如果能联动起来，理论上可以在同一套作业里同时优化“计算并行度”和“资源成本”。不过目前这套联动还需要比较精细的调度策略，我正在做初步验证。

另一个方向是，在数据湖场景下结合AQE并配合依赖关系进一步做查询优化。比如在数据湖的元数据里维护更细粒度的文件级别统计信息，让AQE在shuffle阶段得到的统计信息更准确。升级后的文件统计信息可以反哺给AQE，让动态合并、动态广播策略更精准，进一步减少shuffle数据量。

还有一块我比较关注的是，把AQE的执行经验沉淀成自动化诊断能力。比如从Event Log里解析出倾斜join的检测结果、动态分区合并的前后对比、广播策略是否发生过切换，这些信息汇总之后，可以形成一个“任务体检报告”，让不懂Spark底层原理的同事也能看懂瓶颈在哪，特别适合平台化团队去推进治理工作。

最后再分享一个个人体会。我在最初接触AQE的时候，也走过“开个开关就等着看效果”的弯路，后来发现性能优化这件事，终究还是要回到对业务数据和执行机制的理解上。AQE解决了动态优化的问题，但它不是银弹，它需要你清楚自己的任务哪里慢、为什么慢，然后才能让机制去帮你做最适合的调整。希望这篇实践记录能给正在做Spark性能治理的人一些启发，也欢迎大家多交流各自的调参经验。