漫谈分布式系统（十九）：SQL 的性能黑洞

这是《漫谈分布式系统》系列的第 19 篇，预计会写 30 篇左右。每篇文末有为懒人准备的 TL;DR，还有给勤奋者的关联阅读。扫描文末二维码，关注公众号，听我娓娓道来。也欢迎转发朋友圈分享给更多人。

SQL 里的性能黑洞

SQL 是个历史悠久又非常特殊的语言，在关系型数据库里，我们已经发现并积累了非常多性能优化的经验。

在给 MR 和 Spark 插上 SQL 的翅膀后，我们获得了开发效率上的极大提升，同时，参照关系数据库的经验，应该也能获得了更多性能优化的空间
。

在

这篇文章
里，我们提到，Shuffle 是拖累性能的黑洞，并重点讲了 Spark 在 Shuffle 上做的努力。这些优化效果，在 Spark SQL 里当然会继续发挥作用，比如我们在 SQL 里常写的

group by
。

另外，除了 Shuffle，SQL 里还有另一类操作，也非常影响性能，那就是 Join。
以最常见的两表等值连接为例，示意图是这样：

说白了，就是两张表分别做 shuffle，同时要求把相同的 key 都分到一起，然后匹配筛选出 value 一样的数据。第一步两表分别做 shuffle，之前已经优化过了；第二步匹配数据，很显然也会涉及大量的 IO，很容易成为性能问题的重灾区。
所以对 Join 的优化，直觉上应该首先集中在怎么在 shuffle 分区完之后，更高效地匹配数据。

Nested-Loop Join

以一条查询转账记录的 SQL 为例：

select 

  * 

from 

  sender join transaction 

on 

  sender.id=transaction.sender_id

最直观的办法，就是暴力循环去判断，伪代码是这样：

for i in sender:

  for j in transaction:

    if i.id == j.sender_id:

      ...

这种笛卡尔积


算法的复杂度是 O(m*n)

，m 和 n 分别是两张表的行数。

显然非常低效，没有人会在实际生产中用这种方法。我们甚至反而常常提醒大家，join 一定要带上条件，否则就要算笛卡尔积。

Hash Join

刷过算法题的人很快能反应过来，要比较两个元素是否相等，或者要判断一个元素是否存在，最快的办法就是用 HashTable。也就是所谓 Hash Join，也叫 Shuffled Hash Join。

从上图可见，Shuffled Hash Join 分为三步：

• Shuffle，
• 以其中一张表为基础构建哈希表，
• 遍历另一张表，和哈希表匹配。

可以看到，Hash Join 的


算法复杂度是 O(m+n)

，比笛卡尔积好多了。

本质上来说，Hash Join 充分利用了 Hash 函数的


稳定性

特征，是典型的用


空间换时间

的例子。

但是，当两张表都很大时，空间开销就会很夸张了，甚至可能直接 OOM 挂掉。
我们需要一种不用把一张表的 join key 都放内存的算法。

Sort-Merge Join

但是 Hash Join 已经是 O(m+n) 的复杂度了，只用先后遍历两张表而已。要想匹配，总得遍历一遍吧？
这条路好像行不通。
换个思路，退一步想想，Join 的数据是从哪来的？是 Shuffle 阶段的输出。
而之前 Shuffle 那篇我们刚讲过了，Hive 和 Spark SQL 虽然支持了不少 Shuffle 方式，但大部分情况都会走到 Sort Shuffle。
也就是说，Join 的输入是排序过的。我们要做的是对两个排序过的数组做匹配。

很显然，


不用再遍历整张表了，如果当前行和目标行的 key 不相当，只用继续遍历，直至遇到的 key 比目标行的 key 大为止，因为后面的行都比当前行大，就没有继续遍历下去的必要了

。

Sort Merge Join 不用在内存里构建 Hash Table，省去了大量内存开销，同时能提供和 Hash Join 近似的时间复杂度。
这就使得 Sort Merge Join 能非常高效地处理很多 Join 算法不太擅长的大表和大表的 join 了。
有了 Hash Join 和 Sort-Merge Join，无论表多大，我们都可以很高效地做 join 了。似乎皆大欢喜，可以 happy ending 了。
不过，从 Sort-Merge Join 的例子，我们也可以看到，Shuffle 和 Match 这两个步骤不是割裂的，是会相互影响的。

Sort-Merge Join 正是 Match 受益于 Shuffle 的例子。那反过来，有没有可能 Match 也能对 Shuffle 提供帮助呢
？毕竟，Shuffle 实在太影响性能了。

但是之前我们已经详细讲过 Shuffle 了，好像没有太多优化的空间了。除非…能整个把 Shuffle 去掉…
 

Map Join

去掉 Shuffle 并不是无厘头，join 和 group by 这种操作都需要 shuffle，是为了利用集群扩展性带来的计算能力，分散处理数据，区别在于 join 是多表，而 group by 是单表。

要想对数据分区处理，对一个表 shuffle 已经足以，另一张表其实并非必须 shuffle 不可
。

同时，程序的正确性总是要保证的，所以只能把整张表都发送给另一个表 shuffle 后的所有分区。
很显然，这种办法只适合小表，否则光复制带来的 IO 和构建 HashTable 带来的内存开销就不比 shuffle 低了。
为了进一步提升性能，不对每个分区发送小表，而是对节点或 executor 发送，这样同一个节点或 executor 上的 task 就能复用小表了。
发送完小表后，后面的 join 步骤就和 Hash Join 差不多了。

这个算法应用广泛，在 Spark 里叫 Broadcast Join，在 Hive 里叫 Map Join。
叫 Map Join 是相对于常规的 Hash Join 和 Sort-Merge Join 发生在 Reduce 阶段来说的，Map Join 则在 Map 阶段直接完成数据的关联和筛选。

Bucket Join

Map Join 的思想很好，但只适用大表和小表 join 的情况。如果是大表和大表的 join，还是只能用 Sort-Merge Join。
但是，大表和大表 join 是非常常见的，也是资源消耗大户，如果也能实现 Map Join 的效果，就非常好了。
又要 map join，又不能把整张表广播到各个节点放入内存，唯一的办法就是只广播需要的那部分数据。
需要的那部分数据，很自然想到只广播对应分区的数据。
但是怎么知道一张表的分区，需要另一张表的哪些分区呢？并且是在 join 前事先知道？
只有一种可能，两张表都对 join key 按照相同的方式分区。

于是，从这种方式抽象出了


bucket

的概念。

两张表在写入数据的时候，就都对 join key 按照固定的数量做 bucketing。这样，后续的读操作，就都能自动在 Map 阶段完成 join，而不用 shuffle 到 Reduce 阶段去做 join 了。

Bucket Join
可以看做 bucket 级别的 map join，所以也叫


Bucket Map Join

。

而如果两个表的 bucket 都很大，同样可以参考前面的思路，把匹配数据的过程，由 hash 改为 sort merge，这样就有了所谓


Sort Merge Bucket Join

，或者叫


Sort Merge Bucket Map Join（SMB Map Join）

。

TL;DR

小结下这篇文章的重点：

• Join 是 SQL 里的性能黑洞，有非常大的优化空间，
• 最简单的 Join 方法是 Nested-Loop Join，性能很差，应该尽量避免使用，
• 从传统算法领域借鉴 HashTable 利用空间换时间的办法，有了 Hash Join，
• 大表 join 大表，无法放入内存，考虑到 sort shuffle 已经排序，有了 Sort-Merge Join，
• Sort-Merge Join 是 Join 收益于 Shuffle 的例子，反过来也有优化空间，即 Map Join，直接省略掉 Shuffle，
• 但 Map Join 只适合小表 join 大表，如果大表 join 大表，又不想做 reduce join，只能预先让两表按相同的方式分区，即所谓 Bucket Join，
• Bucket Join 如果使用 hash 的方式匹配数据，就是 Bucket Map Join，如果用 Sort-Merge 的方式，就是 Sort Merge Bucket Map Join。

目前 Hive 和 Spark SQL 都支持了上面主要 Join 中的几种，这里不详细列举了，可以自行查看源码，重点关注 Hive 里的 XXXJoinOperator 类和 Spark SQL 里的 XXXJoinExec 类。
但需要说明的是，Hive 和 Spark SQL 的 join 并不完全兼容。最典型的是 Bucket Join，二者都支持 bucketing，但每个 bucket 的文件数不一样，使用的 hash 算法不一样等。如果想要混用，需要一些改造使得两者兼容。
而到底选择那种 join，可以通知参数设置各种 enable 和阈值来自动选择，也支持通过 hint 的方式强制指定。这篇文章我们主要看思路，具体用法就不再赘述了。
另外值得一提的是，对于很常见的数据倾斜（Data Skew）问题，还有所谓 Skew Join，自动处理。这种 join 不属于通用类 join，这里也不详细介绍了，感兴趣的朋友可以自行了解。
SQL 实在是个太强大的东西，除了 Join 外，还有很多可以优化的地方。
下一篇，我们就一起看下 SQL 的另一个重要优化手段，基于规则的优化（RBO）。