Join 背景介绍
Join 是数据库查询永远绕不开的话题，传统查询 SQL 技术总体可以分为简单操作（过滤操作、排序操作 等），聚合操作-groupby 以及 Join 操作等。其中 Join 操作是最复杂、代价最大的操作类型，也是 OLAP 场景中使用相对较多的操作。
另外，从业务层面来讲，用户在数仓建设的时候也会涉及 Join 使用的问题。通常情况下，数据仓库中的表一般会分为“低层次表”和“高层次表”。
- 所谓“低层次表”：就是数据源导入数仓之后直接生成的表，单表列值较少，一般可以明显归为维度表或事实表，表和表之间大多存在外健依赖，所以查询起来会遇到大量 Join 运算，查询效率很差。
- “高层次表”：是在“低层次表”的基础上加工转换而来，通常做法是使用 SQL 语句将需要 Join 的表预先进行合并形成“宽表”，在宽表上的查询不需要执行大量 Join，效率很高。但宽表缺点是数据会有大量冗余，且相对生成较滞后，查询结果可能并不及时。
为了获得时效性更高的查询结果，大多数场景都需要进行复杂的 Join 操作。Join 操作之所以复杂，主要是通常情况下其时间空间复杂度高，且有很多算法，在不同场景下需要选择特定算法才能获得最好的优化效果。
本文将介绍 SparkSQL 所支持的几种常见的 Join 算法及其适用场景。
基本实现机制
shuffle hash join、broadcast hash join 两者归根到底都属于 hash join，只不过在 hash join 之前需要先 shuffle 还是先 broadcast。其实，hash join 算法来自于传统数据库，而 shuffle 和 broadcast 是大数据的皮（分布式），两者一结合就成了大数据的算法了。因此可以说，大数据的根就是传统数据库。既然 hash join 是“内核”，那就刨出来看看，看完把“皮”再分析一下。
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 在一起。
基本流程可以参考上图，这里有两个小问题需要关注：
1. hash join 性能如何？
- 很显然，hash join 基本都只扫描两表一次，可以认为 o(a+b)，较之最极端的笛卡尔集运算 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，充分利用集群资源并行化。
broadcast hash join
如下图所示，broadcast hash join 可以分为两步：
分步解释：
1.broadcast 阶段：
- 将小表广播分发到大表所在的所有主机。广播算法可以有很多，最简单的是先发给 driver，driver 再统一分发给所有 executor；要不就是基于 BitTorrent 的 TorrentBroadcast。
2. hash join 阶段：
- 在每个 executor 上执行单机版 hash join，小表映射，大表试探。
SparkSQL 规定 broadcast hash join 执行的基本条件为被广播小表必须小于参数 spark.sql.autoBroadcastJoinThreshold，默认为 10M。
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 算法。
看到这里，可以初步总结出来如果两张小表 join 可以直接使用单机版 hash join；如果一张大表 join 一张极小表，可以选择 broadcast hash join 算法；而如果是一张大表 join 一张小表，则可以选择 shuffle hash join 算法；那如果是两张大表进行 join 呢？
sort merge join
SparkSQL 对两张大表 join 采用了全新的算法－sort-merge join，如下图所示
整个过程分为三个步骤：
1. shuffle 阶段：
- 将两张大表根据 join key 进行重新分区，两张表数据会分布到整个集群，以便分布式并行处理。
2. sort 阶段：
- 对单个分区节点的两表数据，分别进行排序。
3. merge 阶段：
- 对排好序的两张分区表数据执行 join 操作。join 操作很简单，分别遍历两个有序序列，碰到相同 join key 就 merge 输出，否则取更小一边。如下图所示：
经过上文的分析，很明显可以得出来这几种 Join 的代价关系：cost(broadcast hash join) < cost(shuffle hash join) < cost(sort merge join)，数据仓库设计时最好避免大表与大表的 join 查询，SparkSQL 也可以根据内存资源、带宽资源适量将参数 spark.sql.autoBroadcastJoinThreshold 调大，让更多 join 实际执行为 broadcast hash join。
参考资料