目录
- 概述
- 动态调整Join策略
- 原理
- 实战
- 动态优化倾斜的 Join
- 原理
- 实战
概述
broadcast hash join 类似于 Spark 共享变量中的广播变量，Spark join 如果能采取这种策略，那join 的性能是最好的
- 自适应查询AQE(Adaptive Query Execution)
- 动态调整Join策略
- 原理
- 实战
- 动态优化倾斜的 Join
- 原理
- 默认环境配置
- 修改配置
- 动态调整Join策略
动态调整Join策略
实际上在生产中，特别是工厂中的局限性，表设计的时候，不是那么合理，导致这这种情况，很少见，很难被调整。
原理
AQE 可以将 sort-merge join 转成 broadcast hash join ，条件是当join 表小于自适应 broadcast hash join 的阀值。
开启了自适应查询执行机制之后，可以在运行时根据最精确的数据指标重新规划join策略，实现动态调整join策略。
看以下图：
后续测试过程中，可以看 spark sql 的执行图。
| 属性名称 | 默认值 | 解释 | 版本 |
|---|---|---|---|
| spark.sql.adaptive.localShuffleReader.enabled | true | 当值为true，且spark.sql.adaptive.enabled也为true时，Spark尝试不需要shuffle分区时，使用本地的shuffle读取器读取shuffle数据，例如：在将 sort-merge 转换成 broadcast-hash join 之后 | 3.0.0 |
| spark.sql.adaptive.autoBroadcastJoinThreshold | (none) | 为表配置最大的字节数，能优化成 broadcast join，通过设置此配置为-1，可以禁用 broadcast ，默认值与 spark.sql.autoBroadcastJoinThreshold 相同 | 3.2.0 |
| spark.sql.autoBroadcastJoinThreshold | 10MB | 同上 | 1.1.0 |
当所有的 shuffle partitions 都小于阀值， AQE 将 sort-merge join 转成 shuffled hash join ；最大阀值配置：spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold
| 属性名称 | 默认值 | 解释 | 版本 |
|---|---|---|---|
| spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold | 0 | 为每个分区配置最大的字节数，能够构建 local hash map,如果这个值不小于 spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold并所有的分区不大于这个配置，join选择更倾向于使用 shuffled hash join，而不是 sort merge join | 3.2.0 |
实战
执行的 sql
select count(*) from xx where dt ='2023-06-30' and workorder='011002118525' ;
## 同样的表相连
select * from (select * from xx where dt ='2023-06-30' and workorder='011002118525') as a
left join xx as b on b.dt ='2023-06-30' and b.workorder='011002118525' and a.id = b.id ;
由上图，三百多万的数据，肯定超过10MB了，所以是 sort merge join
修改 sql 如下：
select * from (select id from xx where dt = '2023-06-30' and workorder='011002118525' ) as a join xx as b on a.id = b.id and b.dt = '2023-06-30' and b.unitid = 'H8TGWJ035ZY0000431';
动态优化倾斜的 Join
原理
数据倾斜严重，将严重影响 join 查询的性能。该功能动态处理在 sort-merge join 倾斜数据时，将其分为大小差不多的任务。当同是启用 spark.sql.adaptive.enabled 和 spark.sql.adaptive.skewJoin.enabled 时，动态优化倾斜 这个功能将生效。
| 属性名称 | 默认值 | 解释 | 版本 |
|---|---|---|---|
| spark.sql.adaptive.skewJoin.enabled | true | 当同是启用 spark.sql.adaptive.enabled，动态优化倾斜 这个功能将生效 | 3.0.0 |
| spark.sql.adaptive.skewJoin.skewedPartitionFactor | 5 | 如果分区的大小大于此因子乘以分区大小的中值，并且也大于spark.sql.adaptive.skewJoin.strakedPartitionThresholdInBytes，则该分区被视为偏斜。 | 3.2.0 |
| spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes | 256MB | 如果分区的字节大小大于此阈值，并且也大于spark.sql.adaptive.skewJoin.strakedPartitionFactor乘以分区大小中值，则该分区被视为偏斜。理想情况下，此配置应设置为大于spark.sql.adaptive.advisoryPartitionSizeInBytes。 | 3.0.0 |
假设有两个表 t1和t2，其中表t1中的P0分区里面的数据量明显大于其他分区，默认的执行情况是这样的，看这个图：
t1表中p0分区的数据比p1\p2\p3这几个分区的数据大很多，可以认为t1表中的数据出现了倾斜。
当t1和t2表中p1、p2、p3这几个分区在join的时候基本上是不会出现数据倾斜的，因为这些分区的数据相对适中。但是P0分区在进行join的时候就会出现数据倾斜了，这样会导致 join 的时间过长。
动态优化倾斜的 join 机制会把P0分区切分成两个子分区P0-1和P0-2，并将每个子分区关联到表t2的对应分区P0，看这个图：
t2表中的P0分区会复制出来两份相同的数据，和t1表中切分出来的P0分区的数据进行 join 关联。
这样相当于就把t1表中倾斜的分区拆分打散了，最终在 join 的时候就不会产生数据倾斜了。
实战
todo: 以后如果遇到，再补充上
