背景
本文基于SPARK 3.3.0
HashAggregate的优化
该优化是FaceBook(Meta) 内部的优化,还有合并到spark社区。
该优化的主要是partialaggregate的部分:对于类似求count,sum,Avg的聚合操作,会存在现在mapper进行部分聚合的操作,之后在reduce端,再进行FinalAggregate操作。这看起来是没有问题的(能够很好的减少网络IO),但是我们知道对于聚合操作,我们会进行数据的spill的操作,如果在mapper阶段合并的数据很少,以至于抵消不了网络IO带来的消耗的话,这无疑会给任务带来损耗。
利用运行时的指标信息,能够达到比较好的加速效果。
ObjectHashAggregate的优化
对于ObjectHashAggreate的原理,可以参考深入理解SPARK SQL 中HashAggregateExec和ObjectHashAggregateExec以及UnsafeRow,该文可以比较清楚的解释ObjectHashAggregate和HashAggregate的区别:
ObjectHashAggregate能够弥补HashAggregate 不能支持collect_set等这种表达式,从而不会转变为SortAggregate
ObjectHashAggregate利用的是java Array对象(SpecificInternalRow)保存聚合的中间缓冲区,这对jvm gc是不太友好的
ObjectHashAggregate根据hashMap的size(默认是128),而不是输入的行数来进行spill,这会导致提前spill,内存利用率不高。
由于提前的spill,ObjectHashAggregate会对剩下的所有数据做额外的一次排序操作(如果没有spill,就不需要额外的sort操作),而HashAggregate则是会对每次需要spill的数据做排序
使用jvm heap的内存使用情况以及处理的行数来指导什么时候开始spill。
但是这种在数据倾斜的情况下,会增加OOM的风险。
SortAggregate优化
目前SortAggreaget的现状是:
每个任务在sort Aggreate前需要按照key进行排序
根据排序的结果,在相邻的行之间进行聚合操作
不同于Hash Aggregate:
不需要hashTable,也就不存在内存溢写和回退到sortAggregate
优化器更喜欢选择hashAggregate
没有codegen的实现.
目前在spark 3.3.0增加的功能:
如果数据是有序的话,会选择用sortAggragate替代HashAggregate
通过物理计划Rule ReplaceHashWithSortAgg 来做替换,当然通过spark.sql.execution.replaceHashWithSortAgg来开启(默认是关闭的),因为对于任何新特性,在release版本默认都是关闭的,在master分支才是开启的
支持sortAggretate(without keys)的codegen代码生成
其他
对于Aggregate更多的细节了解可以参考sparksql源码系列 | 一文搞懂with one count distinct 执行原理