Spark3自适应查询计划（Adaptive Query Execution，AQE）

Posted 2021-08-27 今天努力了吗？？

动态合并shuffle分区（Dynamically coalescing shuffle partitions）
动态调整join策略（Dynamically switching join strategies）
动态优化数据倾斜join（Dynamically optimizing skew joins）
参数：spark.sql.adaptive.enabled 默认关闭，开启此参数后上述三种策略才会执行

1、动态优化数据倾斜（Dynamically optimizing skew joins）
spark.sql.adaptive.skewJoin.enabled 默认 true

相关参数：
①、spark.sql.adaptive.skewJoin.skewedPartitionFactor 默认5
倾斜分区数据大小 > 整个RDD分区分区大小的中位数 * 此参数配置的值

②、spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 默认256MB
倾斜分区数据大小 > 此参数的默认值

同时满足①、②这两个条件参会判定此分区倾斜需要进行裁剪
③、spark.sql.adaptive.advisoryPartitionSizeInBytes 默认64MB
优化后的分区大小 = max(此参数， 非数据倾斜partition的平均大小)

在Reduce阶段进行自动倾斜处理的拆分操作，在同一个Executor内部，本该由一个Task处理的大分区，被AQE拆成多个小分区并交由多个Task去计算，这样可以解决Task之间的负载均衡。但解决不了不同Excuter之间的负载均衡。如果倾斜的分区都分到了一个Executor上，那么这个Executor的计算能力还是整个作业的瓶颈。
如果左右两边的表都出现了数据倾斜现象，需要对左右两张表的倾斜分区都进行拆分操作，左表拆分M各分区，右表拆分N各分区，那么每张表最终需要保证M*N个分区才能保证逻辑关联的一致性。所以在极端情况下对拆分的分区拉取、复制所需要的开销会不可控。
2、动态合并shuffle分区（Dynamically coalescing shuffle partitions）
spark.sql.adaptive.coalescePartitions.enabled 默认 true

优化类型：物理计划 CoalesceShufflePartitions
统计信息：每个Reduce Task分区大小

发生在Shuffle Map完成后的Reduce阶段，Reduce Task将数据分片全部拉回，AQE按照分区编号的顺序，依次把小于目标尺寸的分区合并到一起。目标分区尺寸由一下两个参数决定
spark.sql.adaptive.advisoryPartitionSizeInBytes，默认64M。
spark.sql.adaptive.coalescePartitions.minPartitionNum，最小分区数，默认spark集群的默认并行度。
最终的targetSize为：首先计算出总的shuffle的数据大小totalPostShuffleInputSize；
maxTargetSize为max（totalPostShuffleInputSize/minPartitionNum，16）；targetSize=min（maxTargetSize，advisoryPartitionSizeInBytes）

3、动态调整join策略（Dynamically switching join strategies）
spark.sql.adaptive.localShuffleReader.enabled 默认true

优化类型：逻辑计划 DemoteBroadcastHashJoin
物理计划 OptimizeLocalShuffleReader
统计信息：Map阶段中间文件总大小、中间文件空文件占比

DemoteBroadcastHashJoin：把Shuffle Joins降级为Broadcast Joins。仅适用于Shuffle Sort Merge Join。当两张表完成Shuffle Map阶段后，会继续判断某一张表是否满足一下两个条件
中间文件尺寸总和小于广播阈值 spark.sql.autoBroadcastJoinThreshold(10M)
空文件占比小于配置项 spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin(0.2)
只要有一个表满足就会降级

OptimizeLocalShuffleReader：因为AQE依赖的统计信息来自于Shuffle Map阶段生成的中间文件，所以在AQE开始优化前，Shuffle操作已经执行过半。
两张大表join，超过了广播阈值的话Spark SQL最初会选择SortMerge Join，AQE只有结合两个表join中的Exchange才能进行降级判断，所以两张表必须都完成Map且中间文件落盘。AQE才会决定是否降级以及用那张表做广播变量
spark.sql.adaptive.localShuffleReader.enabled（true）完成省去Shuffle常规操作中的网络分发，Reduce Task可以就读取本地节点（local）的中间文件，完成与广播小表的关联操作。