Spark原理篇之SparkSQL Join分析
1 Join背景
Join是数据库查询永远绕不开的话题，传统查询SQL技术可以分为简单操作（过滤操作-where、排序操作-sort by），聚合操作-groupby以及join操作等。其中join操作是最复杂的、代价最大的操作模型，也是OLAP场景中使用相对较多的操作。因此很有必要对其进行深入研究。
另外，从业务层面来讲，用户在数仓建设的时候也会涉及join使用的问题。通常情况下，数据仓库中的表一般会分为“低层次表”和“高层次表”。
所谓“低层次表”，就是数据源导入数仓之后直接生成的表，单表列值较少，一般可以明显归为维度表（维度表示你要对数据进行分析时所用的一个量, 比如你要分析产品销售情况, 你可以选择按类别来进行分析,或按区域来分析. 这样的按…分析就构成一个维度）或事实表（事实表是数据聚合后依据某个维度生成的结果表），表和表之间大多存在外键依赖，所以查询起来会遇到大量的join运算，查询效率很差。而“高层次表”是在“低层次表”的基础上加工转换而来，通常做法是使用SQL语句将需要join的表预先进行合并形成“宽表”，在宽表上的查询不需要执行大量join，效率很高。但宽表缺点是数据会有大量冗余，且相对生成较滞后，查询结果可能并不及时。
为了获得时效性更高的查询结果，大多数场景都需要进行复杂的join操作。join操作之所以复杂，主要是通常情况下其时间空间复杂度高，且有很多算法，在不同场景下需要选择特定算法才能获得最好的优化效果。
2 Join常见分类以及基本实现机制
当前SparkSQL支持三种join算法：Shuffle Hash Join、Broadcast Hash Join以及Sort Merge Join。其中前两者归根到底都属于Hash Join，只不过载Hash Join之前需要先Shuffle还是先Broadcast。其实，Hash Join算法来自于传统数据库，而Shuffle和Broadcast是大数据的皮（分布式），两者一结合就成了大数据的算法了。因此可以说，大数据的根就是传统数据库。既然Hash Join是“内核”，那就刨出来看看，看完把“皮”再分析一下。
2.1 Hash Join
先来看看这样一条SQL语句：select * from order inner join item on order.i_id=item.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）Shuffle Hash Join
一旦小表的数据量较大，此时就不再适合进行广播分发。这种情况下，可以根据join key相同必然分区相同的原理，将两张表分别按照join key进行重新组织分区，这样就可以将join分而治之，划分为很多小join，充分利用集群资源并行化。
2.1.1 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。
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算法。
看到这里，可以初步总结出来，如果两张小表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操作。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。