spark join 及优化_spark join优化

一 spark常用的join形式：

 

1 broadcast hash join   （小表 大表 join）

1.1）没有加hint,  满足如下条件，也会产生broadcast join：

    1）被广播的表需要小于 spark.sql.autoBroadcastJoinThreshold 所配置的值，如果没有配置，则默认是10M。

     2）被广播的表不能是基表， 比如 left outer join 时，只能广播右表。

     如果将 spark.sql.autoBroadcastJoinThreshold 参数设置为 -1，可以关闭自动 BHJ；

 1.2）加了hint， 只要是等值连接（除full outer join），基本上都会产生broadcast join， 不管参数autoBroadcastJoinThreshold 是否配置。

      eg1:  df1.join(broadcast(df2), $"id1" === $"id2" || $"id2" === $"id3", "left")

      eg2: sparksql:      SELECT /*+ broadcast(a_b) */ * FROM (SELECT /*+ broadcast(a) */ * FROM a JOIN b ON ...) AS a_b JOIN c ON ...

原理： 将小表广播到每个执行器后，构建hash table,    关联有无通过大表关联字段作为key, 从hash表映射即可判断是否有匹配。  相当于遍历一次大表分区即可, 而且分布式运行，性能较好。

 

2 Shuffle Hash Join（SHJ）  （小表 大表 join）

     这种场景适合，小表大于配置的autoBroadcastJoinThreshold=10M,  无法使用broadcast hash join或 小表的数据又不适合使用广播 情形。

     原理：把大表和小表按照相同的分区算法和分区数进行分区（根据参与 Join 的 keys 进行分区），这样就保证了 hash 值一样的不同表数据，在下阶段都分发到的同一个分区中，然后在同一个 Executor 中两张表 hash 值一样的分区就可以在本地进行 hash Join 了。在进行 Join 之前，会对小表 hash 完的分区构建 hash map。

 

    一侧的表要明显小于另外一侧，小的一侧将被广播（明显小于的定义为3倍小，此处为经验值）

 

3  Shuffle Sort Merge Join（SortMergeJoin）     (大表  大表  join)

该JOIN机制是Spark默认的，可以通过参数spark.sql.join.preferSortMergeJoin进行配置，默认是true。

在没有自动识别为broadcast hash join，  默认会使用sort merge join.

 

原理: 也是对两张表参与 Join 的 Keys 使用相同的分区算法和分区数进行分区，保证 hash 值一样的不同表数据，并且每分区排序。在下阶段都分发到同一个分区中，进行 Merge Join。

 

sortmergejoin 会产生 2个表 的  shuffle 和分区内排序。      在join阶段，需要获取2个表的所有task输出，当task数据量大时，具备较大的shuflle和数据读写（每个任务都产生同样数量分区）。

注：适当的情况下可以，在join阶段触发 broadcast hashjoin提升效率，可以一个表的hash分区本地化，另一个表的分区广播到执行器，减少task 数据shuffle和文件读写及连接并发。具体有相关参数控制

 

   4 Broadcast Nested Loop Join

    在没找到合适的连接。      查询语句出现非等值连接时（not in）会使用此方式。

    此方式效率较低。雷士for 循环嵌套比较，    一般不建议使用。 比如2个10w的表，比较次数 10W*10W= 100亿次。 

 

 

 

二 spark 如何选择 join类型：

有join hint基本上按照hint来：

• 1.Broadcast Hint： 如果join类型支持，则选择broadcast hash join
• 2.Sort merge hint：如果join key支持排序的，则选择 sort-merge join
• 3.shuffle hash hint：如果join类型支持 ， 选择 shuffle hash join
• 4.shuffle replicate NL hint：如果是内连接，选择笛卡尔积方式

没有join提示(hints)的情况，则逐个对照下面的规则（源码判断逻辑）

• 1.如果join类型支持，并且其中一张表能够被广播(spark.sql.autoBroadcastJoinThreshold不等于-1（早期版本默认值10MB），最新版本好像默认-1)，且非左连接基表。则选择 broadcast hash join
• 2.如果参数spark.sql.join.preferSortMergeJoin设定为false（默认情况下为true,选择sort-merge join），且一张表足够小(，平均分区小于autoBroadcastJoinThreshold，可以构建一个hash map) ，则选择shuffle hash join。
• 3.如果join keys 是可支持排序的，则选择sort-merge join 。 
• 4.如果是内连接，选择 cartesian join
• 5.如果可能会发生OOM或者没有可以选择的执行策略或非等值连接，则最终选择broadcast nested loop join