一、AQE特性
自适应查询执行（AQE）是Spark SQL中的一种优化技术，它利用运行时统计信息来选择最有效的查询执行计划，也就是说可以根据执行过程中的中间数据优化后续执行，从而提高整体执行效率。核心特征有如下三点：
-
动态设置Shuffle Partition
-
动态调整执行计划
-
数据倾斜优化
1、动态设置Shuffle Partition
1）原有Shuffle的问题
使用Spark SQL时，可通过spark.sql.shuffle.partitions 指定Shuffle时Partition个数，也就是Reducer个数。该参数决定了一个Spark SQL Job中包含的所有Shuffle的Partition个数。如下图所示，当该参数设置为3时，所有Shuffle中Reducer个数都为3。
这种方法存在如下问题：
-
Partition个数不宜设置过大
-
Reducer（代指Spark Shuffle过程中执行Shuffle Read的Task）个数过多，每个Reducer处理的数据量过小。大量小Task造成不必要的task调度开销与可能的资源调度开销（如果开启了Dynamic Allocation）
-
Reducer个数过大，如果Reducer直接写HDFS会生成大量小文件，从而造成大量addBlock RPC，NameNode可能成为瓶颈，并影响其它使用HDFS的应用。
-
过多Reducer写小文件，会造成后面读取这些小文件时产生大量getBlock RPC，对NameNode产生冲击
-
Partition个数不宜设置过小
2）动态设置Shuffle Partition原理
假如Stage1的5个partition数据量分别为60MB，40MB，1MB，2MB，50MB。其中1MB与2MB的Partition明显过小，开启Adaptive Execution后
-
Spark在Stage0的Shuffle Write结束后，根据各Mapper输出，统计得到各partition的数据量，即60MB，40MB，1MB，2MB，50MB
-
通过ExchangeCoordinator计算出合适的post-shuffle Partition个数
-
启动相应个数的Reducer任务
每个Reducer 读取一个或者多个Shuffle Write partition数据（如下图所示，Reducer 0 读取 Partition 0，Reducer 1 读取 Partition 1、2、3，Reducer2读取Partition4）
image
2、动态调整执行计划
在运行时动态调整join策略，在满足条件的情况下，即一张表小于Broadcast阈值，可以讲SortMergeJoin转化为BroadcastHashJoin。
- SortMergeJoin，每个reduce通过网络shuffle读取属于自己的数据，会出现不同程度的数据倾斜问题；
- BroadcastHashJoin，每一个reduce读取一个mapper的整个shuffle output文件，shuffle读变成了本地读取，没有数据通过网络传输；数据量一般比较均匀，也就避免了数据倾斜。
3、数据倾斜优化
在运行时很容易地检测出有数据倾斜的partition，当执行某个stage时，我们收集该stage的每个mapper的shuffle数据大小和记录条数。如果某一个partition数据量或者记录条数超过中位数的N倍，并且大于某个预先配置的阈值，我们就认为这是一个数据倾斜的partition，需要进行特殊的处理。
网友评论