【SparkSQL】聊一聊 Join

╰+攻爆jí腚メ 2023-08-17 17:16 154阅读 0赞

# 1. Join 背景介绍 #

Join 是数据库查询永远绕不开的话题，传统查询 SQL 技术总体可以分为简单操作（过滤操作 WHERE、排序操作 LIMIT 等），聚合操作 GROUPBY 等以及 JOIN 操作等。其中 Join 操作是其中最复杂、代价最大的操作类型，也是 OLAP 场景中使用相对较多的操作。因此很有必要聊聊这个话题。

另外，从业务层面来讲，用户在数仓建设的时候也会涉及 Join 使用的问题。通常情况下，数据仓库中的表一般会分为“低层次表”和“高层次表”。

所谓“低层次表”，就是数据源导入数仓之后直接生成的表，单表列值较少，一般可以明显归为维度表或者事实表，表和表之间大多存在外健依赖，所以查询起来会遇到大量 Join 运算，查询效率相对比较差。而“高层次表”是在”低层次表”的基础上加工转换而来，通常做法是使用 SQL 语句将需要 Join 的表预先进行合并形成“宽表”，在宽表上的查询因为不需要执行大量 Join 因而效率相对较高，很明显，宽表缺点是数据会有大量冗余，而且生成相对比较滞后，查询结果可能并不及时。

因此，为了获得实效性更高的查询结果，大多数场景还是需要进行复杂的 Join 操作。Join 操作之所以复杂，不仅仅因为通常情况下其时间空间复杂度高，更重要的是它有很多算法，在不同场景下需要选择特定算法才能获得最好的优化效果。关系型数据库也有关于 Join 的各种用法，姜承尧大神之前由浅入深地介绍过 MySQL Join 的各种算法以及调优方案。本文接下来会介绍 SparkSQL 所支持的几种常见的 Join 算法以及其适用场景。

# 2. Join 常见分类以及基本实现机制 #

当前 SparkSQL 支持三种 Join 算法

1.  **Shuffle Hash Join**
2.  **Broadcast Hash Join**
3.  **Sort Merge Join**

其中前两者归根到底都属于 Hash Join，只不过在 Hash Join 之前需要先 Shuffle 还是先 Broadcast。其实，这些算法并不是什么新鲜玩意，都是数据库几十年前的老古董了（参考），只不过换上了分布式的皮而已。不过话说回来，SparkSQL/Hive…等等，所有这些大数据技术哪一样不是来自于传统数据库技术，什么语法解析 AST、基于规则优化（CRO）、基于代价优化（CBO）、列存，都来自于传统数据库。就拿 **Shuffle Hash Join** 和 **Broadcast Hash Join** 来说，Hash Join 算法就来自于传统数据库，而 Shuffle 和 Broadcast 是大数据的皮，两者一结合就成了大数据的算法了。因此可以这样说，大数据的根就是传统数据库，传统数据库人才可以很快的转型到大数据。好吧，这些都是闲篇。

继续来看技术，既然 Hash Join 是“内核”，那就刨出来看看，看完把“皮”再分析一下。

## 2.1 Hash Join ##

先来看看这样一条SQL语句：`select * from order,item where item.id = order.i_id`，很简单一个 Join 节点，参与 Join 的两张表是 item 和 order，Join Key 分别是 item.id 以及 order.i\_id。现在假设这个 Join 采用的是 Hash Join 算法，整个过程会经历三步：

1.  确定 Build Table 以及 Probe Table：这个概念比较重要，Build Table 使用 Join Key 构建 Hash Table，而 Probe Table 使用 Join Key 进行探测，探测成功就可以 Join 在一起。通常情况下，小表会作为 Build Table，大表作为 Probe Table。此事例中 item 为Build Table，order 为 Probe Table。
2.  构建 Hash Table：依次读取 Build Table（item）的数据，对于每一行数据根据 Join Key（item.id）进行 Hash，Hash 到对应的 Bucket，生成 Hash Table 中的一条记录。数据缓存在内存中，如果内存放不下需要 dump 到外存。
3.  探测：再依次扫描 Probe Table（order）的数据，使用相同的 Hash 函数映射 Hash Table 中的记录，映射成功之后再检查 Join 条件（item.id = order.i\_id），如果匹配成功就可以将两者 Join 在一起。  
    ![Hash Join][]

基本流程可以参考上图，这里有两个小问题需要关注：

1.  Hash Join 性能如何？很显然，hash join 基本都只扫描两表一次，可以认为 O(a+b)，较之最极端的笛卡尔集运算 O(a\*b)，不知甩了多少条街;
2.  为什么 Build Table 选择小表？道理很简单，因为构建的 Hash Table 最好能全部加载在内存，效率最高；这也决定了 Hash Join 算法只适合至少一个小表的 Join 场景，对于两个大表的 Join 场景并不适用；

上文说过，Hash Join 是传统数据库中的单机 Join 算法，在分布式环境下需要经过一定的分布式改造，说到底就是尽可能利用分布式计算资源进行并行化计算，提高总体效率。Hash Join 分布式改造一般有两种经典方案：

1.  **Broadcast Hash Join**：将其中一张小表广播分发到另一张大表所在的分区节点上，分别并发地与其上的分区记录进行 Hash Join。Broadcast 适用于小表很小，可以直接广播的场景。
2.  **Shuffler Hash Join**：一旦小表数据量较大，此时就不再适合进行广播分发。这种情况下，可以根据 Join Key 相同必然分区相同的原理，将两张表分别按照 Join Key 进行重新组织分区，这样就可以将 Join 分而治之，划分为很多小 Join，充分利用集群资源并行化。

### 2.1.1 Broadcast Hash Join ###

如下图所示，**Broadcast Hash Join** 可以分为两步：

1.  **Broadcast 阶段**：将小表广播分发到大表所在的所有主机。广播算法可以有很多，最简单的是先发给 Driver，Driver 再统一分发给所有 Executor；要不就是基于 Bittorrete 的 P2P 思路；
2.  **Hash Join 阶段**：在每个 Executor 上执行单机版 Hash Join，小表映射，大表试探；  
    ![Broadcast Hash Join][]  
    SparkSQL 规定 **Broadcast Hash Join** 执行的基本条件为被广播小表必须小于参数 spark.sql.autoBroadcastJoinThreshold，默认为10M。

### 2.1.2 Shuffle Hash Join ###

在大数据条件下如果一张表很小，执行 join 操作最优的选择无疑是**Broadcast Hash Join**，效率最高。但是一旦小表数据量增大，广播所需内存、带宽等资源必然就会太大，**Broadcast Hash Join** 就不再是最优方案。此时可以按照 Join Key 进行分区，根据 Key 相同必然分区相同的原理，就可以将大表 Join 分而治之，划分为很多小表的 Join，充分利用集群资源并行化。如下图所示，**Shuffle Hash Join** 也可以分为两步：

1.  **Shuffle 阶段**：分别将两个表按照 Join Key 进行分区，将相同 Join Key 的记录重分布到同一节点，两张表的数据会被重分布到集群中所有节点。这个过程称为 Shuffle；
2.  **Hash Join 阶段**：每个分区节点上的数据单独执行单机 Hash Join 算法。  
    ![Shuffle Hash Join][]  
    看到这里，可以初步总结出来如果两张小表 Join 可以直接使用单机版 Hash Join；如果一张大表 Join 一张极小表，可以选择 **Broadcast Hash Join** 算法；而如果是一张大表 Join 一张小表，则可以选择 **Shuffle Hash Join** 算法；那如果是两张大表进行 Join 呢？

## 2.2 Sort Merge Join ##

SparkSQL 对两张大表 Join 采用了全新的算法 **Sort Merge Join**，如下图所示，整个过程分为三个步骤：

1.  **Shuffle 阶段**：将两张大表根据 Join Key 进行重新分区，两张表数据会分布到整个集群，以便分布式并行处理；
2.  **Sort 阶段**：对单个分区节点的两表数据，分别进行排序；
3.  **Merge 阶段**：对排好序的两张分区表数据执行 Join 操作。  
    ![Sort Merge Join][]  
    Join 操作很简单，分别遍历两个有序序列，碰到相同 Join Key 就 Merge 输出，否则取更小一边，见下图示意：  
    ![Sort Merge Join的 Join 过程][Sort Merge Join_ Join]

仔细分析的话会发现，**Sort Merge Join** 的代价并不比 **Shuffle Hash Join** 小，反而是多了很多。那为什么 SparkSQL 还会在两张大表的场景下选择使用 **Sort Merge Join** 算法呢？这和 Spark 的 Shuffle 实现有关，目前 Spark 的 Shuffle 实现都适用 **Sort Merge Join** 算法，因此在经过 Shuffle 之后 Partition 数据都是按照 Key 排序的。因此理论上可以认为数据经过 Shuffle 之后是不需要 Sort 的，可以直接 Merge。

经过上文的分析，可以明确每种 Join 算法都有自己的适用场景，数据仓库设计时最好避免大表与大表的 Join 查询，SparkSQL 也可以根据内存资源、带宽资源适量将参数 spark.sql.autoBroadcastJoinThreshold 调大，让更多 Join 实际执行为 **Broadcast Hash Join**。

# 3. 总结 #

Join 操作是传统数据库中的一个高级特性，尤其对于当前 MySQL 数据库更是如此，原因很简单，MySQL 对 Join 的支持目前还比较有限，只支持 **Nested Loop Join** 算法，因此在 OLAP 场景下 MySQL 是很难吃的消的，不要去用 MySQL 去跑任何 OLAP 业务，结果真的很难看。不过好消息是 MySQL 在新版本要开始支持 Hash Join 了，这样也许在将来也可以用 MySQL 来处理一些小规模的 OLAP 业务。

和 MySQL 相比，PostgreSQL、SQLServer、Oracle 等这些数据库对 Join 支持更加全面一些，都支持 Hash Join 算法。由 PostgreSQL 作为内核构建的分布式系统 Greenplum 更是在数据仓库中占有一席之地，这和 PostgreSQL 对 Join 算法的支持其实有很大关系。

总体而言，传统数据库单机模式做 Join 的场景毕竟有限，也建议尽量减少使用 Join。然而大数据领域就完全不同，Join 是标配， OLAP 业务根本无法离开表与表之间的关联，对 Join 的支持成熟度一定程度上决定了系统的性能，夸张点说，“得 Join 者得天下”。本文只是试图带大家真正走进 Join 的世界，了解常用的几种 Join 算法以及各自的适用场景。

[Hash Join]: /images/20230808/b6ae62382be14c6dbd24d11e14d78bed.png
[Broadcast Hash Join]: /images/20230808/0285e93c2f6443b08a18fb22408d9538.png
[Shuffle Hash Join]: /images/20230808/14126775850548cd9829aec8371f4007.png
[Sort Merge Join]: /images/20230808/95aa7065c5ed4af9b42f40f100d9f02c.png
[Sort Merge Join_ Join]: /images/20230808/1cbce8722f4448278552e6e70bd44a38.png