引言
join是SQL中的常用操作，良好的表結構能夠將數(shù)據(jù)分散到不同的表中，使其符合某種規(guī)范(mysql三大范式)，可以最大程度的減少數(shù)據(jù)冗余，更新容錯等，而建立表和表之間關系的最佳方式就是join操作。
對于Spark來說有3種Join的實現(xiàn)，每種Join對應的不同的應用場景(SparkSQL自動決策使用哪種實現(xiàn)范式)：
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?，F(xiàn)在假設Join采用的是hash join算法，整個過程會經(jīng)歷三步：
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)的數(shù)據(jù)，對于每一條數(shù)據(jù)根據(jù)Join Key(item.id)進行hash，hash到對應的bucket中(類似于HashMap的原理)，最后會生成一張HashTable，HashTable會緩存在內(nèi)存中，如果內(nèi)存放不下會dump到磁盤中。
3.匹配：生成Hash Table后，在依次掃描Probe Table(order)的數(shù)據(jù)，使用相同的hash函數(shù)(在spark中，實際上就是要使用相同的partitioner)在Hash Table中尋找hash(join key)相同的值，如果匹配成功就將兩者join在一起。
這里有兩個問題需要關注：
1.hash join性能如何？很顯然，hash join基本都只掃描兩表一次，可以認為O(a+b)，較之最極端的是笛卡爾積運算O(a*b)；
2.為什么Build Table選擇小表？道理很簡單，因為構建Hash Table時，最好可以把數(shù)據(jù)全部加載到內(nèi)存中，因為這樣效率才最高，這也決定了hash join只適合于較小的表，如果是兩個較大的表的場景就不適用了。
上文說，hash join是傳統(tǒng)數(shù)據(jù)庫中的單機join算法，在分布式環(huán)境在需要經(jīng)過一定的分布式改造，說到底就是盡可能利用分布式計算資源進行并行計算，提高總體效率，hash join分布式改造一般有以下兩種方案：
1.broadcast hash join：將其中一張較小的表通過廣播的方式，由driver發(fā)送到各個executor，大表正常被分成多個區(qū)，每個分區(qū)的數(shù)據(jù)和本地的廣播變量進行join(相當于每個executor上都有一份小表的數(shù)據(jù)，并且這份數(shù)據(jù)是在內(nèi)存中的，過來的分區(qū)中的數(shù)據(jù)和這份數(shù)據(jù)進行join)。broadcast適用于表很小，可以直接被廣播的場景；
2.shuffle hash join：一旦小表比較大，此時就不適合使用broadcast hash join了。這種情況下，可以對兩張表分別進行shuffle，將相同key的數(shù)據(jù)分到一個分區(qū)中，然后分區(qū)和分區(qū)之間進行join。相當于將兩張表都分成了若干小份，小份和小份之間進行hash join，充分利用集群資源。
Broadcast Hash Join
大家都知道，在數(shù)據(jù)庫的常見模型中(比如星型模型或者雪花模型)，表一般分為兩種：事實表和維度表，維度表一般指固定的、變動較少的表，例如聯(lián)系人、物品種類，一般數(shù)據(jù)有限；而事實表一遍記錄流水，比如銷售清單等，通過隨著時間的增長不斷增長。
因為join操作是對兩個表中key相同的記錄進行連接，在SparkSQL中，對兩個表做join的最直接的方式就是先根據(jù)key進行分區(qū)，再在每個分區(qū)中把key相同的記錄拿出來做連接操作，但這樣不可避免的涉及到shuffle，而shuffle是spark中比較耗時的操作，我們應該盡可能的設計spark應用使其避免大量的shuffle操作。
Broadcast Hash Join的條件有以下幾個：
1.被廣播的表需要小于spark.sql.autoBroadcastJoinThreshold所配置的信息，默認是10M；
2.基表不能被廣播，比如left outer join時，只能廣播右表。
看起來廣播是一個比較理想的方案，但它有沒有缺點呢？缺點也是很明顯的，這個方案只能廣播較小的表，否則數(shù)據(jù)的冗余傳輸就是遠大于shuffle的開銷；另外，廣播時需要被廣播的表collect到driver端，當頻繁的廣播出現(xiàn)時，對driver端的內(nèi)存也是一個考驗。
broadcast hash join可以分為兩步：
1.broadcast階段：將小表廣播到所有的executor上，廣播的算法有很多，最簡單的是先發(fā)給driver，driver再統(tǒng)一分發(fā)給所有的executor，要不就是基于bittorrete的p2p思路；
2.hash join階段：在每個executor上執(zhí)行 hash join，小表構建為hash table，大表的分區(qū)數(shù)據(jù)匹配hash table中的數(shù)據(jù)；
Shuffle Hash Join
當一側的表比較小時，我們可以選擇將其廣播出去以避免shuffle，提高性能。但因為被廣播的表首先被collect到driver端，然后被冗余的發(fā)送給各個executor上，所以當表比較大是，采用broadcast join會對driver端和executor端造成較大的壓力。
我們可以通過將大表和小表都進行shuffle分區(qū)，然后對相同節(jié)點上的數(shù)據(jù)的分區(qū)應用hash join，即先將較小的表構建為hash table，然后遍歷較大的表，在hash table中尋找可以匹配的hash值，匹配成功進行join連接。這樣既在一定程度上減少了driver廣播表的壓力，也減少了executor端讀取整張廣播表的內(nèi)存消耗。
Sshuffle Hash Join分為兩步：
1.對兩張表分別按照join key進行重分區(qū)(分區(qū)函數(shù)相同的時候，相同的相同分區(qū)中的key一定是相同的)，即shuffle，目的是為了讓相同join key的記錄分到對應的分區(qū)中；
2.對對應分區(qū)中的數(shù)據(jù)進行join，此處先將小表分區(qū)構建為一個hash表，然后根據(jù)大表中記錄的join key的hash值拿來進行匹配，即每個節(jié)點山單獨執(zhí)行hash算法。
Shuffle Hash Join的條件有以下幾個：
-
分區(qū)的平均大小不超過spark.sql.autoBroadcastJoinThreshold所配置的值，默認是10M
-
基表不能被廣播，比如left outer join時，只能廣播右表
-
一側的表要明顯小于另外一側，小的一側將被廣播（明顯小于的定義為3倍小，此處為經(jīng)驗值）
看到這里，可以初步總結出來如果兩張小表join可以直接使用單機版hash join；如果一張大表join一張極小表，可以選擇broadcast hash join算法；而如果是一張大表join一張小表，則可以選擇shuffle hash join算法；那如果是兩張大表進行join呢？
Sort Merge Join
上面介紹的方式只對于兩張表有一張是小表的情況適用，而對于兩張大表，但當兩個表都非常大時，顯然無論哪種都會對計算內(nèi)存造成很大的壓力。這是因為join時兩者采取都是hash join，是將一側的數(shù)據(jù)完全加載到內(nèi)存中，使用hash code取join key相等的記錄進行連接。
當兩個表都非常大時，SparkSQL采用了一種全新的方案來對表進行Join，即Sort Merge Join。這種方式不用將一側數(shù)據(jù)全部加載后再進行hash join，但需要在join前將數(shù)據(jù)進行排序。
首先將兩張表按照join key進行重新shuffle，保證join key值相同的記錄會被分在相應的分區(qū)，分區(qū)后對每個分區(qū)內(nèi)的數(shù)據(jù)進行排序，排序后再對相應的分區(qū)內(nèi)的記錄進行連接?？梢钥闯觯瑹o論分區(qū)有多大，Sort Merge Join都不用把一側的數(shù)據(jù)全部加載到內(nèi)存中，而是即用即丟；因為兩個序列都有有序的，從頭遍歷，碰到key相同的就輸出，如果不同，左邊小就繼續(xù)取左邊，反之取右邊。從而大大提高了大數(shù)據(jù)量下sql join的穩(wěn)定性。
SparkSQL對兩張大表join采用了全新的算法－sort-merge join，整個過程分為三個步驟：
-
shuffle階段：將兩張大表根據(jù)join key進行重新分區(qū)，兩張表數(shù)據(jù)會分布到整個集群，以便分布式并行處理；
-
sort階段：對單個分區(qū)節(jié)點的兩表數(shù)據(jù)，分別進行排序；文章來源：http://www.zghlxwxcb.cn/news/detail-676662.html
-
merge階段：對排好序的兩張分區(qū)表數(shù)據(jù)執(zhí)行join操作。join操作很簡單，分別遍歷兩個有序序列，碰到相同join key就merge輸出，否則取更小一邊文章來源地址http://www.zghlxwxcb.cn/news/detail-676662.html
到了這里，關于Spark SQL join的三種實現(xiàn)方式的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！
