拾壹SparkSQL-join:不同的关联形式与实现机制

拾壹SparkSQL-join:不同的关联形式与实现机制

      Join 的种类非常丰富。如果按照关联形式（Join Types）来划分，数据关联分为内关联、外关联、左关联、右关联，等等。对于参与关联计算的两张表，关联形式决定了结果集的数据来源。因此，在开发过程中选择哪种关联形式，是由我们的业务逻辑决定的。而从实现机制的角度，Join 又可以分为 NLJ（Nested Loop Join）、SMJ（Sort Merge Join）和 HJ（Hash Join）。也就是说，同样是内关联，我们既可以采用 NLJ 来实现，也可以采用 SMJ 或是 HJ 来实现。区别在于，在不同的计算场景下，这些不同的实现机制在执行效率上有着天壤之别。因此，了解并熟悉这些机制，对咱们开发者来说至关重要。今天，我们就分别从这两个角度，来说一说 Spark SQL 当中数据关联的来龙去脉。

数据准备

      为了更好地掌握新知识，我会通过一个个例子，为你说明 Spark SQL 数据关联的具体用法。在去介绍数据关联之前，咱们先把示例中会用到的数据准备好。

import spark.implicits._
import org.apache.spark.sql.Dataframe
 
// 创建员工信息表
val seq = Seq((1, "Mike", 28, "Male"), (2, "Lily", 30, "Female")
    , (3, "Raymond", 26, "Male"), (5, "Dave", 36, "Male"))
val employees: Dataframe = seq.toDF("id", "name", "age", "gender")
 
// 创建薪资表
val seq2 = Seq((1, 26000), (2, 30000), (4, 25000), (3, 20000))
val salaries:Dataframe = seq2.toDF("id", "salary")

         我们创建了两个 Dataframe，一个用于存储员工基本信息，我们称之为员工表；另一个存储员工薪水，我们称之为薪资表。数据准备好之后，我们有必要先弄清楚一些数据关联的基本概念。所谓数据关联，它指的是这样一个计算过程：给定关联条件（Join Conditions）将两张数据表以不同关联形式拼接在一起的过程。关联条件包含两层含义，一层是两张表中各自关联字段（Join Key）的选择，另一层是关联字段之间的逻辑关系。在上一讲我们说到，Spark SQL 同时支持 Dataframe 算子与 SQL 查询，因此咱们不妨结合刚刚准备好的数据，分别以这两者为例，来说明数据关联中的基本概念。

 关联形式（Join Types）

      在关联形式这方面，Spark SQL 的支持比较全面，为了让你一上来就建立一个整体的认知，我把 Spark SQL 支持的 Joint Types 都整理到了如下的表格中，你不妨先粗略地过一遍。

 讲下 全外关联的结果集，就是内关联的结果，再加上那些不满足关联条件的左右表剩余数据。要进行全外关联的计算，关键字可以取“full”、“outer”、“fullouter”、或是“full_outer”

val jointDF: Dataframe = salaries.join(employees, 
salaries("id") === employees("id"), "full")
 
jointDF.show
 

左半 / 逆关联（Left Semi Join / Left Anti Join）

       尽管名字听上去拗口，但它们的含义却很简单。我们先来说左半关联，它的关键字有“leftsemi”和“left_semi”。左半关联的结果集，实际上是内关联结果集的子集，它仅保留左表中满足关联条件的那些数据记录，如下表所示。

// 内关联
val jointDF: Dataframe = salaries.join(employees, salaries("id") === 
employees("id"), "inner")
 
jointDF.show
 

 
// 左半关联
val jointDF: Dataframe = salaries.join(employees, salaries("id") === 
employees("id"), "leftsemi")
 
jointDF.show
 

       有了左半关联的基础，左逆关联会更好理解一些。左逆关联同样只保留左表的数据，它的关键字有“leftanti”和“left_anti”。但与左半关联不同的是，它保留的，是那些不满足关联条件的数据记录，如下所示。

// 左逆关联
val jointDF: Dataframe = salaries.join(employees, salaries("id") === employees("id"), "leftanti")
 
jointDF.show
 

       显然，id 为 4 的薪资记录是不满足关联条件 salaries(“id”) === employees(“id”) 的，而左逆关联留下的，恰恰是这些“不达标”的数据记录。好啦，关于 Spark SQL 支持的关联形式，到这里我们就全部说完了。根据这些不同关联形式的特点与作用，再结合实际场景中的业务逻辑，相信你可以在日常的开发中做到灵活取舍。

关联机制（Join Mechanisms）

        不过，从功能的角度出发，使用不同的关联形式来实现业务逻辑，可以说是程序员的一项必备技能。要在众多的开发者中脱颖而出，咱们还要熟悉、了解不同的关联机制。哪怕同样是内关联，不同的 Join 实现机制在执行效率方面差异巨大。因此，掌握不同关联机制的原理与特性，有利于我们逐渐培养出以性能为导向的开发习惯。在本讲的开头，我们提到 Join 有 3 种实现机制，分别是 NLJ（Nested Loop Join）、SMJ（Sort Merge Join）和 HJ（Hash Join）。接下来，我们以内关联为例，结合 salaries 和 employees 这两张表，来说说它们各自的实现原理与特性。

// 内关联
val jointDF: Dataframe = salaries.join(employees, salaries("id") 
=== employees("id"), "inner")
 
jointDF.show
 

NLJ：Nested Loop Join

        对于参与关联的两张表，如 salaries 和 employees，按照它们在代码中出现的顺序，我们约定俗成地把 salaries 称作“左表”，而把 employees 称作“右表”。在探讨关联机制的时候，我们又常常把左表称作是“驱动表”，而把右表称为“基表”。一般来说，驱动表的体量往往较大，在实现关联的过程中，驱动表是主动扫描数据的那一方。而基表相对来说体量较小，它是被动参与数据扫描的那一方。在 NLJ 的实现机制下，算法会使用外、内两个嵌套的 for 循环，来依次扫描驱动表与基表中的数据记录。在扫描的同时，还会判定关联条件是否成立，如内关联例子中的 salaries(“id”) === employees(“id”)。如果关联条件成立，就把两张表的记录拼接在一起，然后对外进行输出。

       在实现的过程中，外层的 for 循环负责遍历驱动表的每一条数据，如图中的步骤 1 所示。对于驱动表中的每一条数据记录，内层的 for 循环会逐条扫描基表的所有记录，依次判断记录的 id 字段值是否满足关联条件，如步骤 2 所示。不难发现，假设驱动表有 M 行数据，而基表有 N 行数据，那么 NLJ 算法的计算复杂度是 O(M * N)。尽管 NLJ 的实现方式简单、直观、易懂，但它的执行效率显然很差。

SMJ：Sort Merge Join

        鉴于 NLJ 低效的计算效率，SMJ 应运而生。Sort Merge Join，顾名思义，SMJ 的实现思路是先排序、再归并。给定参与关联的两张表，SMJ 先把他们各自排序，然后再使用独立的游标，对排好序的两张表做归并关联。

       具体计算过程是这样的：起初，驱动表与基表的游标都会先锚定在各自的第一条记录上，然后通过对比游标所在记录的 id 字段值，来决定下一步的走向。对比结果以及后续 *** 作主要分为 3 种情况：

       满足关联条件，两边的 id 值相等，那么此时把两边的数据记录拼接并输出，然后把驱动表的游标滑动到下一条记录；

      不满足关联条件，驱动表 id 值小于基表的 id 值，此时把驱动表的游标滑动到下一条记录；

      不满足关联条件，驱动表 id 值大于基表的 id 值，此时把基表的游标滑动到下一条记录。

      基于这 3 种情况，SMJ 不停地向下滑动游标，直到某张表的游标滑到尽头，即宣告关联结束。对于驱动表的每一条记录，由于基表已按 id 字段排序，且扫描的起始位置为游标所在位置，因此，SMJ 算法的计算复杂度为 O(M + N)。然而，计算复杂度的降低，仰仗的其实是两张表已经事先排好了序。但是我们知道，排序本身就是一项很耗时的 *** 作，更何况，为了完成归并关联，参与 Join 的两张表都需要排序。因此，SMJ 的计算过程我们可以用“先苦后甜”来形容。苦，指的是要先花费时间给两张表做排序，而甜，指的则是有序表的归并关联能够享受到线性的计算复杂度。

HJ：Hash Join

      考虑到 SMJ 对于排序的苛刻要求，后来又有人推出了 HJ 算法。HJ 的设计初衷是以空间换时间，力图将基表扫描的计算复杂度降低至 O(1)。

        具体来说，HJ 的计算分为两个阶段，分别是 Build 阶段和 Probe 阶段。在 Build 阶段，在基表之上，算法使用既定的哈希函数构建哈希表，如上图的步骤 1 所示。哈希表中的 Key 是 id 字段应用（Apply）哈希函数之后的哈希值，而哈希表的 Value 同时包含了原始的 Join Key（id 字段）和 Payload。

         在 Probe 阶段，算法依次遍历驱动表的每一条数据记录。首先使用同样的哈希函数，以动态的方式计算 Join Key 的哈希值。然后，算法再用哈希值去查询刚刚在 Build 阶段创建好的哈希表。如果查询失败，则说明该条记录与基表中的数据不存在关联关系；相反，如果查询成功，则继续对比两边的 Join Key。如果 Join Key 一致，就把两边的记录进行拼接并输出，从而完成数据关联。

       好啦，到此为止，对于 Join 的 3 种实现机制，我们暂时说到这里。对于它们各自的实现原理，想必你已经有了充分的把握。至于这 3 种机制都适合哪些计算场景，以及 Spark SQL 如何利用这些机制在分布式环境下做数据关联，我们留到下一讲再去展开。

 

对于被连接的数据子集较小的情况，Nested嵌套循环连接是个较好的选择

Hash Join散列连接是CBO 做大数据集连接时的常用方式

SortMergeJoin    通常情况下散列连接的效果都比排序合并连接要好，然而如果行源已经被排过序，在执行排序合并连接时不需要再排序了，这时排序合并连接的性能会优于散列连接。可以使用USE_MERGE(table_name1 table_name2)来强制使用排序合并连接