关于大数据:大数据开发Spark-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 策略的执行过程和实用条件是很有必要的。

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 可排序

当表中的数据比拟大，又不适宜应用播送，这个时候就能够思考应用 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

后面两种 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 可排序

如果 Spark 中两张参加 Join 的表没指定连贯条件，那么会产生 Cartesian product join，这个 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 类型。
吴邪，小三爷，混迹于后盾，大数据，人工智能畛域的小菜鸟。
更多请关注

关于大数据:大数据开发Spark-Join原理详解

1、Broadcast Hash Join

2、Shuffle Hash Join

3、Shuffle Sort Merge Join

4、Cartesian product join

5、Broadcast nested loop join