Spark AQE未按预期生效？开启shuffle分区合并写S3仍生成大量小文件

生效范围的认知偏差 #

• 执行计划中看到AdaptiveSparkPlan isFinalPlan=true仅代表AQE完成了最终执行计划的生成，不代表分区合并逻辑实际触发。AQE包含动态合并分区、动态调整join策略、动态优化shuffle三个独立的优化分支，每个分支的触发都需要单独满足条件。
• AQE的shuffle分区合并逻辑仅作用于shuffle读取阶段的分区，不会对无shuffle依赖的上游map输出分区做无差别合并。如果最终写入Parquet的链路没有经过shuffle环节（比如直接读取CSV后经窄依赖计算就写文件、或广播join后无shuffle直接写），AQE无法获取shuffle分区的大小统计元数据，自然不会触发合并，最终文件数完全由上游输入分片数决定，和AQE开关状态无关。
• 验证方式：在Spark UI的SQL执行页面，查看最终WriteIntoDataSourceCommand节点的上游节点类型，如果节点不是挂在ShuffleQueryStage下，而是直接关联无shuffle的Map阶段、广播阶段，AQE的合并逻辑完全不会介入。

配置项的缺失问题 #

• 目前仅开启了AQE和分区合并的总开关，没有配置小分区的识别阈值参数spark.sql.adaptive.coalescePartitions.minPartitionSize。该参数在Spark 3.x中的默认值为64MB，含义是只有单分区输出小于该值时，才会被判定为小分区并触发合并。如果实际运行中的分区统计值超过该阈值，哪怕分区数很多，也不会进入合并流程。
• 所设置的spark.sql.adaptive.coalescePartitions.minPartitionNum=5是合并后的分区数下限，并非强制要求最终必须生成5个文件：如果全作业总输出大小远小于目标分区大小，AQE会按实际大小生成分区，不会为了凑够5个分区硬拆；如果总输出很大，AQE会按分区大小目标合并，不会强行把分区数降到5。
• 缺失并行度匹配参数spark.sql.adaptive.coalescePartitions.parallelismFirst。该参数默认值为true，意味着AQE会优先按集群CPU核数匹配分区数，而非纯按分区大小做合并，很容易出现大小合并不到位、分区数偏多的情况。

运行模式的干扰 #

• 当前使用spark.master=local[*]的本地运行模式，该模式下Spark对输入分片的大小统计、shuffle分区的元数据采集逻辑和分布式集群存在差异，经常出现分区大小统计不准的问题，导致AQE的阈值判断失效，无法触发合并。本地模式的验证结果不能直接代表生产环境的实际表现。

适配S3存储的修正配置 #

可补全以下参数重新验证效果：

# 保留已开启的基础配置
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.coalescePartitions.minPartitionNum=5
# 补全核心参数
# 小分区判定阈值：小于32MB的分区会被纳入合并范围
spark.sql.adaptive.coalescePartitions.minPartitionSize=32MB
# 关闭优先按CPU数匹配分区的逻辑，纯按目标大小合并分区
spark.sql.adaptive.coalescePartitions.parallelismFirst=false
# 合并后的目标分区大小设为128MB，适配S3存储的最优单文件大小
spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB

如果写入环节上游无shuffle，可在写入前主动通过df.coalesce(目标分区数)或带shuffle的重分区操作，给AQE预留介入空间。

内容的提问来源于stack exchange，提问作者kalyoncu