Spark Join

Join 按照关联形式（Join Types）划分 : 内关联、外关联、左关联、右关联

• Join 按实现机制划分 : NLJ (Nested Loop Join) 、SMJ (Sort Merge Join) 、HJ（Hash Join）
• Join 按分发模式划分 : Shuffle Join、Broadcast Join

关联形式

Spark SQL支持的关联形式 ：

关联形式	Join Type	效果
内关联	inner	结果集中只包含满足关联条件的数据
左外关联	left/leftouter/left_outer	内关联结果集+左表中不满足关联条件的剩余数据
右外关联	right/rightouter/right_outer	内关联结果集 + 右表中不满足关联条件的剩余数据
全外关联	outer/full/fullouter/full_outer	内关联结果集 + 左、右表中不满足关联条件的剩余数据
左半关联	leftsemi/left_semi	内关联结果集，但只保留左表部分的数据
左逆关联	leftanti /left_anti	左表中不满足关联条件的数据

内关联

内关联的效果 : 仅保留左右表中满足关联条件的那些数据记录

• 在员工表与薪资表中，只有 1、2、3 这三个值同时存在它们各自的 id 中。所以结果集中就只有 1、2、3 的这三条数据

// 左表
salaries.show
/** 结果打印
+---+------+
| id|salary|
+---+------+
| 1| 26000|
| 2| 30000|
| 4| 25000|
| 3| 20000|
+---+------+
*/// 右表
employees.show
/** 结果打印
+---+-------+---+------+
| id| name|age|gender|
+---+-------+---+------+
| 1| Mike| 28| Male|
| 2| Lily| 30|Female|
| 3|Raymond| 26| Male|
| 5| Dave| 36| Male|
+---+-------+---+------+
*/// 内关联
val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "inner")jointDF.show
/** 结果打印
+---+------+---+-------+---+------+
| id|salary| id| name|age|gender|
+---+------+---+-------+---+------+
| 1| 26000| 1| Mike| 28| Male|
| 2| 30000| 2| Lily| 30|Female|
| 3| 20000| 3|Raymond| 26| Male|
+---+------+---+-------+---+------+
*/

外关联

外关联能细分 3 种形式：左外关联、右外关联、全外关联

左外关联

左外关联，用 left/ leftouter/ left_outer

• 左外关联的结果集 : 内关联结果集 + 左表的不满足关联条件的剩余数据
• 不存在的记录，在结果集中的所有字段值均为空值 null

val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "left")jointDF.show
/** 结果打印
+---+------+----+-------+----+------+
| id|salary| id| name| age|gender|
+---+------+----+-------+----+------+
| 1| 26000| 1| Mike| 28| Male|
| 2| 30000| 2| Lily| 30|Female|
| 4| 25000|null| null|null| null|
| 3| 20000| 3|Raymond| 26| Male|
+---+------+----+-------+----+------+
*/

右外关联

右外关联，用 right/ rightouter/ right_outer

• 右外关联的结果集：内关联的结果集 + 右表的剩余数据
• 不存在的记录，在结果集中的所有字段值均为空值 null

val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "right")jointDF.show
/** 结果打印
+----+------+---+-------+---+------+
| id|salary| id| name|age|gender|
+----+------+---+-------+---+------+
| 1| 26000| 1| Mike| 28| Male|
| 2| 30000| 2| Lily| 30|Female|
| 3| 20000| 3|Raymond| 26| Male|
|null| null| 5| Dave| 36| Male|
+----+------+---+-------+---+------+
*/

全外关联

全外关联，用 full/ outer/ ullouter/ full_outer

• 全外关联的结果集：内关联的结果 + 那些不满足关联条件的左右表剩余数据

val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "full")jointDF.show
/** 结果打印
+----+------+----+-------+----+------+
| id|salary| id| name| age|gender|
+----+------+----+-------+----+------+
| 1| 26000| 1| Mike| 28| Male|
| 3| 20000| 3|Raymond| 26| Male|
|null| null| 5| Dave| 36| Male|
| 4| 25000|null| null|null| null|
| 2| 30000| 2| Lily| 30|Female|
+----+------+----+-------+----+------+
*/

左半/逆关联

左半关联，用 leftsemi/left_semi

• 左半关联的结果集 : 内关联结果集的子集，但仅保留左表数据

// 左半关联
val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "left_semi")jointDF.show
/** 结果打印
+---+------+
| id|salary|
+---+------+
| 1| 26000|
| 2| 30000|
| 3| 20000|
+---+------+
*/

左逆关联，用 leftanti/left_anti

• 左逆关联的结果集 : 不满足条件结果集的子集，但仅保留左表数据

// 左逆关联
val jointDF: DataFrame = salaries.join(employees, salaries("id") === employees("id"), "left_anti")jointDF.show
/** 结果打印
+---+------+
| id|salary|
+---+------+
| 4| 25000|
+---+------+
*/

关联机制

Join 有 3 种实现机制 :

• NLJ（Nested Loop Join）: 嵌套循环连接
• SMJ（Sort Merge Join）: 排序归并连接
• HJ（Hash Join）: 哈希连接

俗定 : 左表 = 驱动表，右表 = 基表

• 驱动表较大，主动扫描数据的一边
• 基表较小，被动参与数据扫描的一方

Join实现机制	范围	效率	工作原理
Nested Loop Join	全部关联	最差	用嵌套循环来实现关联，效率最低，算法复杂度为 O(M * N)
Sort Merge Join	等值关联	次优	先将两表排序，再用游标滑动实现关联，算法复杂度为 O(M + N)
Hash Join	等值关联	最优	关联过程分两阶段：Build：用哈希算法对基表建立哈希表。Probe：遍历驱动表每条数据，动态计算哈希值，再找哈希表来实现关联计算。复杂度为 O(M)

NLJ

NLJ (Nested Loop Join ) 的实现机制：用外、内两个嵌套的 for 循环，来依次扫描驱动表与基表中的数据记录

• 外层的 for 循环遍历驱动表的每一条数据
• 驱动表中的每条数据，内层 for 逐条扫描基表的所有记录，依次判断记录的 id 字段值是否满足关联条件
• 驱动表有 M 行，基表有 N 行，NLJ 计算复杂度是 O(M * N)

SMJ

SMJ (Sort Merge Join) 的实现思路 : 先排序、再归并

• 对关联的两张表，SMJ 先各自排序，然后再使用独立的游标，对排好序的两张表做归并关联
• SMJ 算法的计算复杂度为 O(M + N)

游标对比的 3 种情况：

• 满足关联条件：两边的 id 相等，把两边的数据记录拼接并输出，然后驱动表的游标下滑
• 不满足关联条件：驱动表 id 值 < 基表的 id 值，驱动表的游标下滑
• 不满足关联条件 : 驱动表 id 值 > 基表的 id 值，基表的游标下滑

HJ

HJ (Hash Join) 的设计初衷 : 以空间换时间，将基表的计算复杂度降到 O(1)

HJ 的计算的两个阶段：Build 阶段和 Probe 阶段

• Build 阶段：在基表上，用自定的哈希构建哈希表。哈希表的 Key 是 id 哈希后的哈希值，哈希表的 Value 是基表数据
• Probe 阶段：依次遍历驱动表的每条数据。先用同样的哈希，得到哈希值。然后用哈希值去查询刚 Build 好的哈希表。当查询失败，就跳过；当查询成功，就对比两边的 Join Key。如果 Join Key 一致，就拼接并输出

分发模式

Join 按照分发模式划分 : Shuffle Join、Broadcast Join

• Shuffle Join ：任何情况，都能完成数据关联的计算
• Broadcast Join : 广播数据表的全量数据到 Driver 的内存、以及各个 Executors 的内存

Join策略	前提条件	优势	劣势
Shuffle Join	无	适用范围广，不受数据体量、内存大小	会有 l/O开销，容易性能瓶颈
Broadcast Join	基表 < Executors 内存	只需广播基表，消除驱动表的 Shuffle 过程，执行效率高	无

用 Shuffle 完成数据关联 :

用广播机制完成数据关联 :

6 种分布式 Join :

Spark SQL 的5 种 Join ：

Join 选择

关联条件	Join 策略排序
等值关联	Broadcast HJ > Shuffle SMJ > Shuffle HJ
不等值关联	Broadcast NLJ > Shuffle NLJ

等值 Join

等值数据关联时，Spark 会按照 BHJ > SMJ > SHJ 的顺序选择 Join 策略

BHJ 效率最高，前提条件：

• 连接类型不能是全连接（Full Outer Join）
• 基表要足够小，能放到广播变量

SHJ 前提条件：

• 外表大小大于内表的 3 倍上
• 内表数据分片的平均大小 < 广播变量阈值

spark.sql.join.preferSortMergeJoin为 False 时，Spark SQL 才会先尝试 SHJ

不等值 Join

不等值 Join 只能用 BNLJ和 CPJ

• Spark SQL 会按照 BNLJ > CPJ 的顺序尝试
• BNLJ 前提条件：内表小能放进广播变量