引言
join是SQL中的常用操作，良好的表結構能夠將數據分散到不同的表中，使其符合某種規范(mysql三大范式)，可以最大程度的減少數據冗余，更新容錯等，而建立表和表之間關系的最佳方式就是join操作。
對於Spark來說有3種Join的實現，每種Join對應的不同的應用場景(SparkSQL自動決策使用哪種實現范式)：
1.Broadcast Hash Join：適合一張很小的表和一張大表進行Join；
2.Shuffle Hash Join：適合一張小表(比上一個大一點)和一張大表進行Join；
2.Sort Merge Join：適合兩張大表進行Join；
前兩者都是基於Hash Join的，只不過Hash Join之前需要先shuffle還是先brocadcast。下面詳細解釋一下這三種Join的具體原理。
Hash Join
先來看看這樣一條SQL語句：select * from order,item where item.id = order.i_id，參與join的兩張表是order和item，join key分別是item.id以及order.i_id。現在假設Join采用的是hash join算法，整個過程會經歷三步：
1.確定Build Table以及Probe Table：這個概念比較重要，Build Table會被構建成以join key為key的hash table，而Probe Table使用join key在這張hash table表中尋找符合條件的行，然后進行join鏈接。Build表和Probe表是Spark決定的。通常情況下，小表會被作為Build Table，較大的表會被作為Probe Table。
2.構建Hash Table：依次讀取Build Table(item)的數據，對於每一條數據根據Join Key(item.id)進行hash，hash到對應的bucket中(類似於HashMap的原理)，最后會生成一張HashTable，HashTable會緩存在內存中，如果內存放不下會dump到磁盤中。
3.匹配：生成Hash Table后，在依次掃描Probe Table(order)的數據，使用相同的hash函數(在spark中，實際上就是要使用相同的partitioner)在Hash Table中尋找hash(join key)相同的值，如果匹配成功就將兩者join在一起。
基礎流程可以參考上圖，這里有兩個問題需要關注：
1.hash join性能如何？很顯然，hash join基本都只掃描兩表一次，可以認為O(a+b)，較之最極端的是笛卡爾積運算O(a*b)；
2.為什么Build Table選擇小表？道理很簡單，因為構建Hash Table時，最好可以把數據全部加載到內存中，因為這樣效率才最高，這也決定了hash join只適合於較小的表，如果是兩個較大的表的場景就不適用了。
上文說，hash join是傳統數據庫中的單機join算法，在分布式環境在需要經過一定的分布式改造，說到底就是盡可能利用分布式計算資源進行並行計算，提高總體效率，hash join分布式改造一般有以下兩種方案：
1.broadcast hash join：將其中一張較小的表通過廣播的方式，由driver發送到各個executor，大表正常被分成多個區，每個分區的數據和本地的廣播變量進行join(相當於每個executor上都有一份小表的數據，並且這份數據是在內存中的，過來的分區中的數據和這份數據進行join)。broadcast適用於表很小，可以直接被廣播的場景；
2.shuffle hash join：一旦小表比較大，此時就不適合使用broadcast hash join了。這種情況下，可以對兩張表分別進行shuffle，將相同key的數據分到一個分區中，然后分區和分區之間進行join。相當於將兩張表都分成了若干小份，小份和小份之間進行hash join，充分利用集群資源。
Broadcast Hash Join
大家都知道，在數據庫的常見模型中(比如星型模型或者雪花模型)，表一般分為兩種：事實表和維度表，維度表一般指固定的、變動較少的表，例如聯系人、物品種類，一般數據有限；而事實表一遍記錄流水，比如銷售清單等，通過隨着時間的增長不斷增長。
因為join操作是對兩個表中key相同的記錄進行連接，在SparkSQL中，對兩個表做join的最直接的方式就是先根據key進行分區，再在每個分區中把key相同的記錄拿出來做連接操作，但這樣不可避免的涉及到shuffle，而shuffle是spark中比較耗時的操作，我們應該盡可能的設計spark應用使其避免大量的shuffle操作。
當維度表和事實表進行Join操作時，為了避免shuffle，我們可以將大小有限的維度表的全部數據分發到每個節點上，供事實表使用。executor存儲維度表的全部數據，一定程度上犧牲了空間，換取shuffle操作大量的耗時，這在SparkSQL中稱作Broadcast Join，如下圖所示：
Table B是較小的表，黑色表示將其廣播到每個executor節點上，Table A的每個partition會通過block manager取到Table A的數據。根據每條記錄的Join Key取到Table B中相對應的記錄，根據Join Type進行操作。這個過程比較簡單，不做贅述。
Broadcast Hash Join的條件有以下幾個：
1.被廣播的表需要小於spark.sql.autoBroadcastJoinThreshold所配置的信息，默認是10M；
2.基表不能被廣播，比如left outer join時，只能廣播右表。
看起來廣播是一個比較理想的方案，但它有沒有缺點呢？缺點也是很明顯的，這個方案只能廣播較小的表，否則數據的冗余傳輸就是遠大於shuffle的開銷；另外，廣播時需要被廣播的表collect到driver端，當頻繁的廣播出現時，對driver端的內存也是一個考驗。
如下圖所示，broadcast hash join可以分為兩步：
1.broadcast階段：將小表廣播到所有的executor上，廣播的算法有很多，最簡單的是先發給driver，driver再統一分發給所有的executor，要不就是基於bittorrete的p2p思路；
2.hash join階段：在每個executor上執行 hash join，小表構建為hash table，大表的分區數據匹配hash table中的數據；
Shuffle Hash Join
當一側的表比較小時，我們可以選擇將其廣播出去以避免shuffle，提高性能。但因為被廣播的表首先被collect到driver端，然后被冗余的發送給各個executor上，所以當表比較大是，采用broadcast join會對driver端和executor端造成較大的壓力。
我們可以通過將大表和小表都進行shuffle分區，然后對相同節點上的數據的分區應用hash join，即先將較小的表構建為hash table，然后遍歷較大的表，在hash table中尋找可以匹配的hash值，匹配成功進行join連接。這樣既在一定程度上減少了driver廣播表的壓力，也減少了executor端讀取整張廣播表的內存消耗。
Sshuffle Hash Join分為兩步：
1.對兩張表分別按照join key進行重分區(分區函數相同的時候，相同的相同分區中的key一定是相同的)，即shuffle，目的是為了讓相同join key的記錄分到對應的分區中；
2.對對應分區中的數據進行join，此處先將小表分區構建為一個hash表，然后根據大表中記錄的join key的hash值拿來進行匹配，即每個節點山單獨執行hash算法。
Shuffle Hash Join的條件有以下幾個：
1. 分區的平均大小不超過spark.sql.autoBroadcastJoinThreshold所配置的值，默認是10M
2. 基表不能被廣播，比如left outer join時，只能廣播右表
3. 一側的表要明顯小於另外一側，小的一側將被廣播（明顯小於的定義為3倍小，此處為經驗值）
看到這里，可以初步總結出來如果兩張小表join可以直接使用單機版hash join；如果一張大表join一張極小表，可以選擇broadcast hash join算法；而如果是一張大表join一張小表，則可以選擇shuffle hash join算法；那如果是兩張大表進行join呢？
Sort Merge Join
上面介紹的方式只對於兩張表有一張是小表的情況適用，而對於兩張大表，但當兩個表都非常大時，顯然無論哪種都會對計算內存造成很大的壓力。這是因為join時兩者采取都是hash join，是將一側的數據完全加載到內存中，使用hash code取join key相等的記錄進行連接。
當兩個表都非常大時，SparkSQL采用了一種全新的方案來對表進行Join，即Sort Merge Join。這種方式不用將一側數據全部加載后再進行hash join，但需要在join前將數據進行排序。
首先將兩張表按照join key進行重新shuffle，保證join key值相同的記錄會被分在相應的分區，分區后對每個分區內的數據進行排序，排序后再對相應的分區內的記錄進行連接。可以看出，無論分區有多大，Sort Merge Join都不用把一側的數據全部加載到內存中，而是即用即丟；因為兩個序列都有有序的，從頭遍歷，碰到key相同的就輸出，如果不同，左邊小就繼續取左邊，反之取右邊。從而大大提高了大數據量下sql join的穩定性。
SparkSQL對兩張大表join采用了全新的算法－sort-merge join，如下圖所示，整個過程分為三個步驟：
. shuffle階段：將兩張大表根據join key進行重新分區，兩張表數據會分布到整個集群，以便分布式並行處理；
2. sort階段：對單個分區節點的兩表數據，分別進行排序；
3. merge階段：對排好序的兩張分區表數據執行join操作。join操作很簡單，分別遍歷兩個有序序列，碰到相同join key就merge輸出，否則取更小一邊，見下圖示意：
參考：
https://www.cnblogs.com/0xcafedaddy/p/7614299.html
https://www.cnblogs.com/0xcafedaddy/p/7614299.html