Spark 3.0 新特性 之 自适应查询与分区动态裁剪

Spark憋了一年半的大招后，发布了3.0版本，新特性主要与Spark SQL和Python相关。这也恰恰说明了大数据方向的两大核心：BI与AI。下面是本次发布的主要特性，包括性能、API、生态升级、数据源、SQL兼容、监控和调试等方面的升级。

本次主要整理了性能方面的优化，包括了自适应查询与动态分区裁剪。

1 自适应查询

AQE，Adaptive Query Execution，说的简单点就是让Spark在运行中根据搜集到的信息灵活采取优化手段，提升性能。

说起这个可以先回想下Spark的发展历史，在1.x时代Spark通过RDD的编程形成DAG图，这个阶段可以说没啥优化完全是按照规则来执行；在2.x时代，引入了代价计算，Spark会通过提前进行代价计算，选择代价最小的查询计划（跟大部分的数据库类似，代价计算依赖于数据本身的统计，如数据量、文件大小、分区数等，由于Spark是存储与计算分离的模式，因此这些统计信息有时候会缺失或者不准确，那么得到的查询代价自然也就不准确了）；在3.x时代，引入自适应查询，即在运行的过程中可以根据得到的缓存数据信息动态调整分区策略、join策略等。这样就保证了刚开始表的统计信息不准，可能查询计划不是最高效的，但是随着查询的执行，可以动态优化整个查询计划。

那么到底自适应都可以做什么呢？

1.1 动态分区合并

在Spark的经典优化策略里，调整分区数从而改变并行度是最基本的优化手段，可以调整的分区数却不是那么容易找到最优值的。分区数太小，可能导致单个分区内的数据太多，单个任务的执行效率低下；分区数太大，可能导致碎片太多，任务之间来回切换浪费性能。比如经典的shuffle操作后，每个shuffle数据都需要对应的reduce端接收处理，如果分区数过多，有可能导致某几个任务读取的数据量很小，造成资源的浪费。

引入AQE后，Spark会自动把数据量很小的分区进行合并处理：

1.2 动态join策略选择

在Spark中支持多种join策略，这些策略在不同的分布式框架中差不多。分别是：

• Broadcast Hash Join（BHJ），广播 join
• Shuffle Hash Join（SHJ），哈希 join
• Sort Merge Join（SMJ），排序 join

BHJ是当小表与大表关联时，把小表广播到大表的每个分区中，每个分区都与完整的小表进行关联，最后合并得到结果。像Spark会配置一个参数 spark.sql.autoBroadcastJoinThreshold 来决定小于这个配置的表就认为是小表，然后采用广播策略（默认10MB）。一般广播的套路是把小表拷贝到driver端，然后分发到每个executor工作节点上，因此如果表的数据太大，会导致来回复制的数据太多，性能低下，因此BHJ仅适用于广播小表。

SHJ是针对表的数据量过大时，按照分区列进行打散，两张表按照不同的分区重新排列数据。不过这种JOIN方法也有个弊端，就是需要对应分区的两张表数据都同时加载完成，才能开始计算。如果两张表的数据量都很大，有可能会造成分区节点内存溢出。

SMJ是针对上述的情况，在确定shuffle分区后对数据进行排序，这样两张表可以不需要等待数据全部加载到内存，只要对应的排序数据部分加载完成后就可以提前开始。

总结完三种join策略后，可以发现假设由于数据统计信息的缺失或不准确，或者是过滤条件的影响，可能会按照原来表的大小判断join的策略。比如某个表初始的时候15M，达不到广播join的要求，但是该表在查询过程中有个filter条件可以让表仅保留8M的有效数据，此时就可以采用广播join了。AQE就是利用这种特性，在运行时动态检测表的大小，当表的大小达到要求后会优化join为广播join。

1.3 数据倾斜优化

在分布式查询中某个查询任务会同时分拆成多个任务运行在不同的机器上，假设某个任务对应的数据量很大，就会引发数据倾斜的问题。比如下面的两张表关联,但是左表的第一个分区数据量很多,就会引发数据倾斜问题.

AQE可以在运行时检测到数据倾斜,并把大分区分割成多个小分区同时与对应的右表进行关联。

2 动态分区裁剪

这个比较好理解，正常Spark或Hive在查询时，会根据查询条件与分区字段自动过滤底层的数据文件。但是如果过滤条件没有及时的反映到查询上，就会导致数据被冗余加载。比如左边的是没有动态分区裁剪的情况，两张表进行关联操作，左表包含一个过滤条件，右表需要全表读取。经过动态分区优化后，右表可以直接添加过滤条件，如 id in (select id from lefttable where filter_cond) , 这样可以提前过滤掉部分数据。

3 关联提示

之前在Flink中看到过这种用法，即在sql中使用某种代码提示，让编译器根据代码提示选择优化策略执行。语法如：/** xxx /。比如 select /* BROADCAST(a) */ * from a join b on a.id = b.id，可以强制a表广播与b表进行关联操作。

以上就是主要的性能方面的优化。其他方面由于工作内容涉及的不多，因此就先不过多整理了，感兴趣可以去官网或者观看上面的分享视频。需要额外一提的是，官方文档也有两个很重要的调整：

1 增加了SQL相关的文档

2 增加了UI方面的说明

后续会分享更多Spark相关的原理和特性文章。