等值Join VS 非等值Join
- SparkSQL和HiveSQL不同，HiveSQL只支持等值连接，但是SparkSQL非等值连接也是支持的。
- 等值连接和非等值连接的区别是：如果on语句中包含一个相等条件或多个需要同时满足的相等条件，那么称为等值连接，否则就称为非等值连接。
- 非等值连接例如下面这两种：A.x < B.x A.x == B.x or A.y == B.y，除非业务需求，否者尽量不要使用非等值连接，相对等值连接要慢。
Join类型
数据分析中将两个数据集进行 Join 操作是很常见的场景。在 Spark 的物理计划阶段，Spark 的 Join Selection 类会根据 Join hints 策略、Join 表的大小、 Join 是等值 Join 还是不等值以及参与 Join 的 key 是否可以排序等条件来选择最终的 Join 策略，最后 Spark 会利用选择好的 Join 策略执行最终的计算。当前 Spark 一共支持五种 Join 策略：
- Broadcast hash join (BHJ)
- Shuffle hash join（SHJ）
- Shuffle sort merge join (SMJ)
- Shuffle-and-replicate nested loop join，又称笛卡尔积（Cartesian product join)
- Broadcast nested loop join (BNLJ)
其中 BHJ和 SMJ这两种 Join 策略是我们运行 Spark 作业最常见的。JoinSelection会先根据 Join的 Key 为等值 Join来选择Broadcast hash join、Shuffle hash join 以及Shuffle sort merge join 中的一个；如果 Join 的 Key 为不等值Join 或者没有指定 Join 条件，则会选择 Broadcast nested loop join 或 Shuffle-and-replicate nested loop join。
不同的 Join 策略在执行上效率差别很大，了解每种 Join 策略的执行过程和适用条件是很有必要的。
1、Broadcast Hash Join
Broadcast Hash Join 的实现是将小表的数据广播到 Spark所有的 Executor端，这个广播过程和我们自己去广播数据没什么区别：
- 利用 collect 算子将小表的数据从 Executor 端拉到 Driver 端
- 在 Driver 端调用 sparkContext.broadcast 广播到所有 Executor 端
- 在 Executor 端使用广播的数据与大表进行 Join 操作（实际上是执行map操作）
这种 Join 策略避免了 Shuffle 操作。一般而言，Broadcast Hash Join 会比其他 Join 策略执行的要快。
使用这种 Join 策略必须满足以下条件：
- 小表的数据必须很小，可以通过 spark.sql.autoBroadcastJoinThreshold参数来配置，默认是 10MB
- 如果内存比较大，可以将阈值适当加大
- 将 spark.sql.autoBroadcastJoinThreshold参数设置为 -1，可以关闭这种连接方式
- 只能用于等值 Join，不要求参与 Join 的 keys 可排序
Broadcast hash Join原理
一般根据数据表的角色不同分为streamedTable流式表和BuildTable构建表，通常会把大表设定为流式表，将小表设定为构建表（TableB）。在Join运算过程中，会遍历流式表的每条记录，然后在构建表中查找相匹配的记录进行匹配。BHJ这种Join方式会将小表Broadcast到各个Executor上，构建成“HashMap”，然后大表从“HashMap”中寻找对应的记录关联到一起，这样就规避掉了Shuffle（分布式场景下shuffle是非常耗费带宽和内存的）。流程图如下所示：
2、Shuffle Hash Join
当表中的数据比较大，又不适合使用广播，这个时候就可以考虑使用 Shuffle Hash Join。
Shuffle Hash Join同样是在大表和小表进行 Join 的时候选择的一种策略。它的计算思想是：把大表和小表按照相同的分区算法和分区数进行分区（根据参与 Join 的 keys 进行分区），这样就保证了 hash 值一样的数据都分发到同一个分区中，然后在同一个 Executor 中两张表 hash 值一样的分区就可以在本地进行 hash Join 了。在进行 Join 之前，还会对小表的分区构建 Hash Map。Shuffle hash join利用了分治思想，把大问题拆解成小问题去解决。
要启用 Shuffle Hash Join必须满足以下条件：
- 仅支持等值 Join，不要求参与 Join 的 Keys 可排序
- spark.sql.join.preferSortMergeJoin 参数必须设置为 false，参数是从 Spark 2.0.0 版本引入的，默认值为true，也就是默认情况下选择 Sort Merge Join
- 小表的大小（plan.stats.sizeInBytes）必须小于 spark.sql.autoBroadcastJoinThreshold *spark.sql.shuffle.partitions（默认值200）
- 而且小表大小（stats.sizeInBytes）的三倍必须小于等于大表的大小（stats.sizeInBytes），也就是a.stats.sizeInBytes * 3 < = b.stats.sizeInBytes
Shuffle hash Join原理
和上面的差不多，最后也是对构建表构造一个“HashMap”。不过它是使用ZipPartitions的方式将两个RDD中数据中相同Key的数据划分到同一个分区中，然后对小表中的数据构建HashMap。如果数据量大的话，这个HashMap占用的内存也不小，所以说这种Join方式是对内存有要求的（一般在spark里面用的较少，可手动打开）。
3、Shuffle Sort Merge Join
前面两种 Join 策略对表的大小都有条件的，如果参与 Join 的表都很大，这时候就得考虑用 Shuffle Sort Merge Join了。
Shuffle Sort Merge Join 的实现思想：
- 将两张表按照 join key 进行shuffle，保证join key值相同的记录会被分在相应的分区
- 对每个分区内的数据进行排序
- 排序后再对相应的分区内的记录进行连接
无论分区有多大，Sort Merge Join都不用把一侧的数据全部加载到内存中，而是即用即丢；因为两个序列都有序。从头遍历，碰到key相同的就输出，如果不同，左边小就继续取左边，反之取右边。从而大大提高了大数据量下sql join的稳定性。
要启用 Shuffle Sort Merge Join 必须满足以下条件：
- 仅支持等值 Join，并且要求参与 Join的 Keys 可排序
Shuffle Sort Merge Join原理
当两个表的数据量都非常大时，会使用SortMergeJoin方式进行Join。也是对两张表进行ZipPartition操作，将相同Key的数据划分到同一个分区中，但是之后不会把构建表构造HashMap了，而是对分区中的数据按照Join Key 排序，然后对排序之后的数据不停迭代，按照Key的顺序逐个对比寻找匹配的记录。流程图如下所示：
可以看到，首先将两张表按照join keys进行了重新shuffle，保证join keys值相同的记录会被分在相应的分区。分区后对每个分区内的数据进行排序，排序后再对相应的分区内的记录进行连接。因为两个序列都是有序的，从头遍历，碰到key相同的就输出；如果不同，左边小就继续取左边，反之取右边。
可以看出，无论分区有多大，Sort Merge Join都不用把某一侧的数据全部加载到内存中，而是即用即取即丢，从而大大提升了大数据量下sql join的稳定性。
4、Cartesian product join
如果 Spark 中两张参与 Join的表没指定连接条件，那么会产生 Cartesian product join，这个 Join 得到的结果其实就是两张表行数的乘积。
5、Broadcast nested loop join
可以把 Broadcast nested loop join 的执行看做下面的计算：
for record_1 in relation_1:
for record_2 in relation_2:
join condition is executed
可以看出 Broadcast nested loop join 在某些情况会对某张表重复扫描多次，效率非常低下。从名字可以看出，这种join 会根据相关条件对小表进行广播，以减少表的扫描次数。
Broadcast nested loop join 支持等值和不等值 Join，支持所有的 Join 类型。
总结
分布式join常用的join方式就是如下3种，其他的分布式计算组件（clickhouse、flink、dorisdb，startrocks，presto等）的join实现原理几乎就是蜕变自这3种，深入掌握这三种分布式join的原理对理解其他分布式组件的join原理非常有必要。
- Broadcast hash join (BHJ)、
- Shuffle hash join（SHJ）、
- Shuffle sort merge join (SMJ)
分布式join基本的优化方式（应该是对其他组件都适用的方式）
- 最好是低基数表join起来更快，低基数表放右边
- check自己的sql，看join的两张表是否能够做谓词下推（别让查询优化器帮你做，有的查询优化器没那么智能），减少join表，特别是右表的数据量和条数
- 如果两张大表join，建议可以使用spark这种引擎来辅助计算，像clickhouse这种纯内存的计算引擎耗时久，极度耗费资源，影响adhoc查询响应，计算失败是要重头开始算的。当然也可以优化的，比如clickhouse里面的colocate join可以帮助优化，具体根据自己场景来选择
后续还会介绍下clickhouse的join实现方式，尽请期待。