Spark3.0 新特性

spark2.0 的 cost-based optimizer 优化器

• CBO 需要对表进行统计分析, 但因为 spark 的数据源往往很大, 导致代价高昂; 且往往数据的变动, 导致数据的分布是不可预测的; 其次, UDF这种黑盒计算也无法做出优化

基于以上原因, AQE 使用一种简单的思想:
(1) optimizer: 用RBO对sql进行执行计划的优化
(2) planner: 随着执行计划的每一步完成, 都会带来更精准的统计信息, 用这些实时产生的统计信息优化查询

有了AQE, 就能完成3中潜在的目标
We can convert the soft merge join to broadcast hash join, based on the runtime statistics.
(1) 查看 shuffle operators 产生的统计信息, 如果发现产生的数据很少, 就自动把 soft merge join 计划变为 broadcast hash join 计划

(2) 2.0 时代, 一个常用的性能调优配置是 spark.sql.shuffle.partitions, 默认值是 200 . 这是一个全局配置, 并不能使用所有查询. 如果这个值配置的太小, 会导致单个分区文件过大; 则在将数据合并排序输出时需要更大的磁盘IO; 如果这个数配置的太大, 则会导致分区数量太多, 造成IO效率不足, 而且主要的瓶颈会在 task scheduler 上

We can shrink the number of reducers after over-partitioning.
(3) Dynamically Coalesce Shuffle Partitions
对于上面的问题(2), 如果能动态合并 shuffle partition, 则可以对 spark.sql.shuffle.partitions 设置一个很大的初始值, 当执行完上一个 stage 后, 就可以知道每个 partition 的精确大小, 就可以自动和合并( coalesce )相邻的 partition , 从而自动降低 partition 的个数为一个很小的值
We can also handle the skew join at runtime.
把存在数据倾斜的 partition 经过 split 后, 才形成一个个小的 subpartition . 比如表 A 有一个很大的分区 A0 (存在数据倾斜), 该分区 A0 需要和 表B 的分区 B0 join 产生数据, 则3.0 会先把 A0 split 成3个小的子分区, 并拷贝分区 B0 为3份, 从而让 A0 的三个子分区并行执行 shuffle reading, sorting, merging, 同时也避免 sort merge join 产生出一个巨大的分区

SELECT t1.id, t2.pKey 
FROM t1
JOIN t2
ON t1.pKey = t2.pKey
AND t2.id < 2