前言：

寫SQL的時候很多時候都有用到join語句，但是我們真的有仔細想過數據在join的過程到底是怎么樣的嗎？今天借這位大神的文章來交接下sql中join的原理。同樣，如有冒犯，請聯系。

Join背景介紹

Join是數據庫查詢永遠繞不開的話題，傳統查詢SQL技術總體可以分為簡單操作（過濾操作-where、排序操作-limit等），聚合操作-groupBy等以及Join操作等。其中Join操作是其中最復雜、代價最大的操作類型，也是OLAP場景中使用相對較多的操作。因此很有必要聊聊這個話題。
另外，從業務層面來講，用戶在數倉建設的時候也會涉及Join使用的問題。通常情況下，數據倉庫中的表一般會分為”低層次表”和“高層次表”。
所謂”低層次表”，就是數據源導入數倉之后直接生成的表，單表列值較少，一般可以明顯歸為維度表或者事實表，表和表之間大多存在外健依賴，所以查詢起來會遇到大量Join運算，查詢效率相對比較差。而“高層次表”是在”低層次表”的基礎上加工轉換而來，通常做法是使用SQL語句將需要Join的表預先進行合并形成“寬表”，在寬表上的查詢因為不需要執行大量Join因而效率相對較高，很明顯，寬表缺點是數據會有大量冗余，而且生成相對比較滯后，查詢結果可能并不及時。
因此，為了獲得實效性更高的查詢結果，大多數場景還是需要進行復雜的Join操作。Join操作之所以復雜，不僅僅因為通常情況下其時間空間復雜度高，更重要的是它有很多算法，在不同場景下需要選擇特定算法才能獲得最好的優化效果。關系型數據庫也有關于Join的各種用法，姜承堯大神之前由淺入深地介紹過MySQL Join的各種算法以及調優方案（關注公眾號InsideMySQL并回復join可以查看相關文章）。本文接下來會介紹SparkSQL所支持的幾種常見的Join算法以及其適用場景。

Join常見分類以及基本實現機制

**當前SparkSQL支持三種Join算法－shuffle hash join、broadcast hash join以及sort merge join。**其中前兩者歸根到底都屬于hash join，只不過在hash join之前需要先shuffle還是先broadcast。其實，這些算法并不是什么新鮮玩意，都是數據庫幾十年前的老古董了（參考），只不過換上了分布式的皮而已。不過話說回來，SparkSQL/Hive…等等，所有這些大數據技術哪一樣不是來自于傳統數據庫技術，什么語法解析AST、基于規則優化（CRO）、基于代價優化（CBO）、列存，都來自于傳統數據庫。就拿shuffle hash join和broadcast hash join來說，hash join算法就來自于傳統數據庫，而shuffle和broadcast是大數據的皮，兩者一結合就成了大數據的算法了。因此可以這樣說，大數據的根就是傳統數據庫，傳統數據庫人才可以很快的轉型到大數據。好吧，這些都是閑篇。
繼續來看技術，既然hash join是’內核’，那就刨出來看看，看完把’皮’再分析一下。

Hash Join

先來看看這樣一條SQL語句：

select * from order,item where item.id = order.i_id，

很簡單一個Join節點，參與join的兩張表是item和order，join key分別是item.id以及order.i_id。現在假設這個Join采用的是hash join算法，整個過程會經歷三步：
1、 確定Build Table以及Probe Table：這個概念比較重要，Build Table使用join key構建Hash Table，而Probe Table使用join key進行探測，探測成功就可以join在一起。通常情況下，小表會作為Build Table，大表作為Probe Table。此事例中item為Build Table，order為Probe Table。
2、 構建Hash Table：依次讀取Build Table（item）的數據，對于每一行數據根據join key（item.id）進行hash，hash到對應的Bucket，生成hash table中的一條記錄。數據緩存在內存中，如果內存放不下需要dump到外存。（這里是先利用join key hash到對應的bucket中，然后利用相同的hash規則去連接另一張表中相同的key數據）
3、探測：再依次掃描Probe Table（order）的數據，使用相同的hash函數映射Hash Table中的記錄，映射成功之后再檢查join條件（item.id = order.i_id），如果匹配成功就可以將兩者join在一起。
總結：
確定小表為build table,大表為probe table；之后利用小表的join key構建hash table。掃描小表全表，不同的join key被分發到不同的bucket下，之后再依次掃描probe table，按照同樣的hash規則，將數據hash到不同的bucket下，之后在bucket下掃描小表數據，若join條件一致，則可將兩者join在一起。

基本流程可以參考上圖，這里有兩個小問題需要關注：
1、 hash join性能如何？很顯然，hash join基本都只掃描兩表一次，可以認為o(a+b)，較之最極端的笛卡爾集運算a*b，不知甩了多少條街。
2、為什么Build Table選擇小表？道理很簡單，因為構建的Hash Table最好能全部加載在內存，效率最高；這也決定了hash join算法只適合至少一個小表的join場景，對于兩個大表的join場景并不適用；
上文說過，hash join是傳統數據庫中的單機join算法，在分布式環境下需要經過一定的分布式改造，說到底就是盡可能利用分布式計算資源進行并行化計算，提高總體效率。hash join分布式改造一般有兩種經典方案：
1、broadcast hash join：將其中一張小表廣播分發到另一張大表所在的分區節點上，分別并發地與其上的分區記錄進行hash join。broadcast適用于小表很小，可以直接廣播的場景。
2、shuffler hash join：
一旦小表數據量較大，此時就不再適合進行廣播分發。這種情況下，可以根據join key相同必然分區相同的原理，將兩張表分別按照join key進行重新組織分區，這樣就可以將join分而治之，劃分為很多小join，充分利用集群資源并行化。（相當于在map端將大小按照key進行拆分重新組織分區，然后根據key分發到reduce端進行分別大小表的處理，最終再將結果進行匯總。）

Broadcast Hash Join

如下圖所示，broadcast hash join可以分為兩步：
1、broadcast階段：將小表廣播分發到大表所在的所有主機。廣播算法可以有很多，最簡單的是先發給driver，driver再統一分發給所有executor；要不就是基于bittorrete的p2p思路；
基于bittorrete的p2p思路可參考：

https://zhidao.baidu.com/question/9782615.html
https://baike.baidu.com/item/BitTorrent/142795?fr=aladdin

2、hash join階段：在每個executor上執行單機版hash join，小表映射，大表試探；
SparkSQL規定broadcast hash join執行的基本條件為被廣播小表必須小于參數spark.sql.autoBroadcastJoinThreshold，默認為10M。

Shuffle Hash Join

在大數據條件下如果一張表很小，執行join操作最優的選擇無疑是broadcast hash join，效率最高。但是一旦小表數據量增大，廣播所需內存、帶寬等資源必然就會太大，broadcast hash join就不再是最優方案。此時可以按照join key進行分區，根據key相同必然分區相同的原理，就可以將大表join分而治之，劃分為很多小表的join，充分利用集群資源并行化。如下圖所示，shuffle hash join也可以分為兩步：
1、shuffle階段：分別將兩個表按照join key進行分區，將相同join key的記錄重分布到同一節點，兩張表的數據會被重分布到集群中所有節點。這個過程稱為shuffle
2、hash join階段：每個分區節點上的數據單獨執行單機hash join算法。（最后應該還要做一個union all的操作將之前處理的內容進行合并）

看到這里，**可以初步總結出來如果兩張小表join可以直接使用單機版hash join；如果一張大表join一張極小表，可以選擇broadcast hash join算法；**而如果是一張大表join一張小表，則可以選擇shuffle hash join算法；那如果是兩張大表進行join呢？

Sort-Merge Join

SparkSQL對兩張大表join采用了全新的算法－sort-merge join，如下圖所示，整個過程分為三個步驟：

1、shuffle階段：將兩張大表根據join key進行重新分區，兩張表數據會分布到整個集群，以便分布式并行處理
2、sort階段：對單個分區節點的兩表數據，分別進行排序

3、merge階段：對排好序的兩張分區表數據執行join操作。join操作很簡單，分別遍歷兩個有序序列，碰到相同join key就merge輸出，否則取更小一邊（兩張分區表進行join的過程中，會不斷的比較索引的大小，一直以索較小的索引值遍歷分區表數據。），見下圖示意：

仔細分析的話會發現，sort-merge join的代價并不比shuffle hash join小，反而是多了很多。那為什么SparkSQL還會在兩張大表的場景下選擇使用sort-merge join算法呢？這和Spark的shuffle實現有關，目前spark的shuffle實現都適用sort-based shuffle算法，因此在經過shuffle之后partition數據都是按照key排序的。因此理論上可以認為數據經過shuffle之后是不需要sort的，可以直接merge（也就是說sort-merge-join實際只需要執行shuffle和merge階段，而shuffle-hash-join需要執行shuffle和hash-join階段。而對于大表join大表來說，merge階段比hash-join階段更優！
為什么更優：hash-join的復雜度O(a+b)；而merge小于O(a+b)。a,b代表數組的長度）。
經過上文的分析，可以明確每種Join算法都有自己的適用場景，數據倉庫設計時最好避免大表與大表的join查詢，SparkSQL也可以根據內存資源、帶寬資源適量將參數spark.sql.autoBroadcastJoinThreshold調大，讓更多join實際執行為broadcast hash join。

總結

Join操作是傳統數據庫中的一個高級特性，尤其對于當前MySQL數據庫更是如此，原因很簡單，MySQL對Join的支持目前還比較有限，只支持Nested-Loop Join算法，因此在OLAP場景下MySQL是很難吃的消的，不要去用MySQL去跑任何OLAP業務，結果真的很難看。不過好消息是MySQL在新版本要開始支持Hash Join了，這樣也許在將來也可以用MySQL來處理一些小規模的OLAP業務。
和MySQL相比，PostgreSQL、SQLServer、Oracle等這些數據庫對Join支持更加全面一些，都支持Hash Join算法。由PostgreSQL作為內核構建的分布式系統Greenplum更是在數據倉庫中占有一席之地，這和PostgreSQL對Join算法的支持其實有很大關系。
總體而言，傳統數據庫單機模式做Join的場景畢竟有限，也建議盡量減少使用Join。然而大數據領域就完全不同，Join是標配，OLAP業務根本無法離開表與表之間的關聯，對Join的支持成熟度一定程度上決定了系統的性能，夸張點說，’得Join者得天下’。本文只是試圖帶大家真正走進Join的世界，了解常用的幾種Join算法以及各自的適用場景。

參考：http://hbasefly.com/2017/03/19/sparksql-basic-join/