网站导航
> 文章列表 > Spark Join

Spark Join

网友: 文章列表 2024-03-21 16:05:25

Spark Join

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)

Spark Join

SMJ

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

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

游标对比的 3 种情况：

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

Spark Join

HJ

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

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

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

Spark Join

分发模式

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

  • Shuffle Join ：任何情况，都能完成数据关联的计算
  • Broadcast Join : 广播数据表的全量数据到 Driver 的内存、以及各个 Executors 的内存
Join策略 前提条件 优势 劣势
Shuffle Join 无 适用范围广，不受数据体量、内存大小 会有 l/O开销，容易性能瓶颈
Broadcast Join 基表 < Executors 内存 只需广播基表，消除驱动表的 Shuffle 过程，执行效率高 无

用 Shuffle 完成数据关联 :

Spark Join

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

Spark Join

6 种分布式 Join :

Spark Join

Spark SQL 的5 种 Join ：

Spark 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 前提条件：内表小能放进广播变量
网络标签：

相关问题