Spark Join
- 关联形式
- 内关联
- 外关联
- 左外关联
- 右外关联
- 全外关联
- 左半/逆关联
- 关联机制
- NLJ
- SMJ
- HJ
- 分发模式
- Join 选择
- 等值 Join
- 不等值 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 前提条件：内表小能放进广播变量
