數(shù)據(jù)分析中將兩個數(shù)據(jù)集進(jìn)行 Join 操作是很常見的場景。在 Spark 的物理計劃階段，Spark 的 Join Selection 類會根
據(jù) Join hints 策略、Join 表的大小、 Join 是等值 Join 還是不等值以及參與 Join 的 key 是否可以排序等條件來選擇最
終的 Join 策略，最后 Spark 會利用選擇好的 Join 策略執(zhí)行最終的計算。當(dāng)前 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 策略是我們運(yùn)行 Spark 作業(yè)最常見的。JoinSelection 會先根據(jù) 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 策略在執(zhí)行上效率差別很大，了解每種 Join 策略的執(zhí)行過程和適用條件是很有必要的。

1、Broadcast Hash Join

Broadcast Hash Join 的實現(xiàn)是將小表的數(shù)據(jù)廣播到 Spark 所有的 Executor 端，這個廣播過程和我們自己去廣播數(shù)
據(jù)沒什么區(qū)別：

利用 collect 算子將小表的數(shù)據(jù)從 Executor 端拉到 Driver 端
在 Driver 端調(diào)用 sparkContext.broadcast 廣播到所有 Executor 端
在 Executor 端使用廣播的數(shù)據(jù)與大表進(jìn)行 Join 操作（實際上是執(zhí)行map操作）

這種 Join 策略避免了 Shuffle 操作。一般而言，Broadcast Hash Join 會比其他 Join 策略執(zhí)行的要快。

使用這種 Join 策略必須滿足以下條件：
小表的數(shù)據(jù)必須很小，可以通過 spark.sql.autoBroadcastJoinThreshold 參數(shù)來配置，默認(rèn)是 10MB
如果內(nèi)存比較大，可以將閾值適當(dāng)加大
將 spark.sql.autoBroadcastJoinThreshold 參數(shù)設(shè)置為 -1，可以關(guān)閉這種連接方式
只能用于等值 Join，不要求參與 Join 的 keys 可排序

2、Shuffle Hash Join

當(dāng)表中的數(shù)據(jù)比較大，又不適合使用廣播，這個時候就可以考慮使用 Shuffle Hash Join。
Shuffle Hash Join 同樣是在大表和小表進(jìn)行 Join 的時候選擇的一種策略。它的計算思想是：把大表和小表按照相同
的分區(qū)算法和分區(qū)數(shù)進(jìn)行分區(qū)（根據(jù)參與 Join 的 keys 進(jìn)行分區(qū)），這樣就保證了 hash 值一樣的數(shù)據(jù)都分發(fā)到同一
個分區(qū)中，然后在同一個 Executor 中兩張表 hash 值一樣的分區(qū)就可以在本地進(jìn)行 hash Join 了。在進(jìn)行 Join 之
前，還會對小表的分區(qū)構(gòu)建 Hash Map。Shuffle hash join 利用了分治思想，把大問題拆解成小問題去解決。

要啟用 Shuffle Hash Join 必須滿足以下條件：
僅支持等值 Join，不要求參與 Join 的 Keys 可排序
spark.sql.join.preferSortMergeJoin 參數(shù)必須設(shè)置為 false，參數(shù)是從 Spark 2.0.0 版本引入的，默認(rèn)值為
true，也就是默認(rèn)情況下選擇 Sort Merge Join
小表的大小（plan.stats.sizeInBytes）必須小于 spark.sql.autoBroadcastJoinThreshold *
spark.sql.shuffle.partitions（默認(rèn)值200）
而且小表大?。╯tats.sizeInBytes）的三倍必須小于等于大表的大小（stats.sizeInBytes），也就是
a.stats.sizeInBytes * 3 < = b.stats.sizeInBytes

3、Shuffle Sort Merge Join

前面兩種 Join 策略對表的大小都有條件的，如果參與 Join 的表都很大，這時候就得考慮用 Shuffle Sort Merge Join
了。
Shuffle Sort Merge Join 的實現(xiàn)思想：
將兩張表按照 join key 進(jìn)行shuffle，保證join key值相同的記錄會被分在相應(yīng)的分區(qū)
對每個分區(qū)內(nèi)的數(shù)據(jù)進(jìn)行排序
排序后再對相應(yīng)的分區(qū)內(nèi)的記錄進(jìn)行連接
無論分區(qū)有多大，Sort Merge Join都不用把一側(cè)的數(shù)據(jù)全部加載到內(nèi)存中，而是即用即丟；因為兩個序列都有序。從
頭遍歷，碰到key相同的就輸出，如果不同，左邊小就繼續(xù)取左邊，反之取右邊。從而大大提高了大數(shù)據(jù)量下sql join
的穩(wěn)定性。

要啟用 Shuffle Sort Merge Join 必須滿足以下條件：

僅支持等值 Join，并且要求參與 Join 的 Keys 可排序

4、Cartesian product join

如果 Spark 中兩張參與 Join 的表沒指定連接條件，那么會產(chǎn)生 Cartesian product join，這個 Join 得到的結(jié)果其實

就是兩張表行數(shù)的乘積。

5、Broadcast nested loop join

可以把 Broadcast nested loop join 的執(zhí)行看做下面的計算：

for record_1 in relation_1:

for record_2 in relation_2:

join condition is executed

可以看出 Broadcast nested loop join 在某些情況會對某張表重復(fù)掃描多次，效率非常低下。從名字可以看出，這種

join 會根據(jù)相關(guān)條件對小表進(jìn)行廣播，以減少表的掃描次數(shù)。

Broadcast nested loop join 支持等值和不等值 Join，支持所有的 Join 類型。