Join背景介紹

Join是數(shù)據(jù)庫(kù)查詢永遠(yuǎn)繞不開(kāi)的話題,，傳統(tǒng)查詢SQL技術(shù)總體可以分為簡(jiǎn)單操作（過(guò)濾操作-where,、排序操作-limit等），聚合操作-groupBy等以及Join操作等,。其中Join操作是其中最復(fù)雜,、代價(jià)最大的操作類(lèi)型，也是OLAP場(chǎng)景中使用相對(duì)較多的操作,。因此很有必要聊聊這個(gè)話題,。

另外，從業(yè)務(wù)層面來(lái)講,，用戶在數(shù)倉(cāng)建設(shè)的時(shí)候也會(huì)涉及Join使用的問(wèn)題,。通常情況下，數(shù)據(jù)倉(cāng)庫(kù)中的表一般會(huì)分為”低層次表”和“高層次表”,。

所謂”低層次表”,，就是數(shù)據(jù)源導(dǎo)入數(shù)倉(cāng)之后直接生成的表，單表列值較少,，一般可以明顯歸為維度表或者事實(shí)表,，表和表之間大多存在外健依賴(lài)，所以查詢起來(lái)會(huì)遇到大量Join運(yùn)算,，查詢效率相對(duì)比較差,。而“高層次表”是在”低層次表”的基礎(chǔ)上加工轉(zhuǎn)換而來(lái)，通常做法是使用SQL語(yǔ)句將需要Join的表預(yù)先進(jìn)行合并形成“寬表”,，在寬表上的查詢因?yàn)椴恍枰獔?zhí)行大量Join因而效率相對(duì)較高,，很明顯，寬表缺點(diǎn)是數(shù)據(jù)會(huì)有大量冗余,，而且生成相對(duì)比較滯后,，查詢結(jié)果可能并不及時(shí)。

因此,，為了獲得實(shí)效性更高的查詢結(jié)果,，大多數(shù)場(chǎng)景還是需要進(jìn)行復(fù)雜的Join操作。Join操作之所以復(fù)雜,，不僅僅因?yàn)橥ǔＧ闆r下其時(shí)間空間復(fù)雜度高,，更重要的是它有很多算法，在不同場(chǎng)景下需要選擇特定算法才能獲得最好的優(yōu)化效果,。關(guān)系型數(shù)據(jù)庫(kù)也有關(guān)于Join的各種用法,，姜承堯大神之前由淺入深地介紹過(guò)MySQL Join的各種算法以及調(diào)優(yōu)方案（關(guān)注公眾號(hào)InsideMySQL并回復(fù)join可以查看相關(guān)文章）。本文接下來(lái)會(huì)介紹SparkSQL所支持的幾種常見(jiàn)的Join算法以及其適用場(chǎng)景,。

Join常見(jiàn)分類(lèi)以及基本實(shí)現(xiàn)機(jī)制

當(dāng)前SparkSQL支持三種Join算法－shuffle hash join,、broadcast hash join以及sort merge join。其中前兩者歸根到底都屬于hash join,，只不過(guò)在hash join之前需要先shuffle還是先broadcast,。其實(shí)，這些算法并不是什么新鮮玩意,，都是數(shù)據(jù)庫(kù)幾十年前的老古董了（參考）,，只不過(guò)換上了分布式的皮而已。不過(guò)話說(shuō)回來(lái),，SparkSQL/Hive…等等,，所有這些大數(shù)據(jù)技術(shù)哪一樣不是來(lái)自于傳統(tǒng)數(shù)據(jù)庫(kù)技術(shù)，什么語(yǔ)法解析AST,、基于規(guī)則優(yōu)化（CRO）,、基于代價(jià)優(yōu)化（CBO）、列存,，都來(lái)自于傳統(tǒng)數(shù)據(jù)庫(kù),。就拿shuffle hash join和broadcast hash join來(lái)說(shuō)，hash join算法就來(lái)自于傳統(tǒng)數(shù)據(jù)庫(kù),，而shuffle和broadcast是大數(shù)據(jù)的皮,，兩者一結(jié)合就成了大數(shù)據(jù)的算法了。因此可以這樣說(shuō),，大數(shù)據(jù)的根就是傳統(tǒng)數(shù)據(jù)庫(kù),，傳統(tǒng)數(shù)據(jù)庫(kù)人才可以很快的轉(zhuǎn)型到大數(shù)據(jù)。好吧,，這些都是閑篇,。

繼續(xù)來(lái)看技術(shù)，既然hash join是’內(nèi)核’,，那就刨出來(lái)看看,，看完把’皮’再分析一下。

Hash Join

先來(lái)看看這樣一條SQL語(yǔ)句：select * from order,item where item.id = order.i_id,，很簡(jiǎn)單一個(gè)Join節(jié)點(diǎn),，參與join的兩張表是item和order,，join key分別是item.id以及order.i_id。現(xiàn)在假設(shè)這個(gè)Join采用的是hash join算法,，整個(gè)過(guò)程會(huì)經(jīng)歷三步：

1. 確定Build Table以及Probe Table：這個(gè)概念比較重要,，Build Table使用join key構(gòu)建Hash Table，而Probe Table使用join key進(jìn)行探測(cè),，探測(cè)成功就可以join在一起,。通常情況下，小表會(huì)作為Build Table,，大表作為Probe Table,。此事例中item為Build Table，order為Probe Table,。

2. 構(gòu)建Hash Table：依次讀取Build Table（item）的數(shù)據(jù),，對(duì)于每一行數(shù)據(jù)根據(jù)join key（item.id）進(jìn)行hash，hash到對(duì)應(yīng)的Bucket,，生成hash table中的一條記錄,。數(shù)據(jù)緩存在內(nèi)存中，如果內(nèi)存放不下需要dump到外存,。

3. 探測(cè)：再依次掃描Probe Table（order）的數(shù)據(jù),，使用相同的hash函數(shù)映射Hash Table中的記錄，映射成功之后再檢查join條件（item.id = order.i_id）,，如果匹配成功就可以將兩者join在一起,。

基本流程可以參考上圖，這里有兩個(gè)小問(wèn)題需要關(guān)注：

1. hash join性能如何,？很顯然,，hash join基本都只掃描兩表一次，可以認(rèn)為o(a+b),，較之最極端的笛卡爾集運(yùn)算a*b,，不知甩了多少條街

2. 為什么Build Table選擇小表？道理很簡(jiǎn)單,，因?yàn)闃?gòu)建的Hash Table最好能全部加載在內(nèi)存,，效率最高；這也決定了hash join算法只適合至少一個(gè)小表的join場(chǎng)景,，對(duì)于兩個(gè)大表的join場(chǎng)景并不適用,；

上文說(shuō)過(guò)，hash join是傳統(tǒng)數(shù)據(jù)庫(kù)中的單機(jī)join算法,，在分布式環(huán)境下需要經(jīng)過(guò)一定的分布式改造,，說(shuō)到底就是盡可能利用分布式計(jì)算資源進(jìn)行并行化計(jì)算，提高總體效率。hash join分布式改造一般有兩種經(jīng)典方案：

1. broadcast hash join：將其中一張小表廣播分發(fā)到另一張大表所在的分區(qū)節(jié)點(diǎn)上,，分別并發(fā)地與其上的分區(qū)記錄進(jìn)行hash join,。broadcast適用于小表很小，可以直接廣播的場(chǎng)景,。

2. shuffler hash join：一旦小表數(shù)據(jù)量較大,，此時(shí)就不再適合進(jìn)行廣播分發(fā)。這種情況下,，可以根據(jù)join key相同必然分區(qū)相同的原理，將兩張表分別按照join key進(jìn)行重新組織分區(qū),，這樣就可以將join分而治之,，劃分為很多小join，充分利用集群資源并行化,。

Broadcast Hash Join

如下圖所示,，broadcast hash join可以分為兩步：

1. broadcast階段：將小表廣播分發(fā)到大表所在的所有主機(jī)。廣播算法可以有很多,，最簡(jiǎn)單的是先發(fā)給driver,，driver再統(tǒng)一分發(fā)給所有executor；要不就是基于bittorrete的p2p思路,；

2. hash join階段：在每個(gè)executor上執(zhí)行單機(jī)版hash join,，小表映射，大表試探,；

SparkSQL規(guī)定broadcast hash join執(zhí)行的基本條件為被廣播小表必須小于參數(shù)spark.sql.autoBroadcastJoinThreshold,，默認(rèn)為10M。

Shuffle Hash Join

在大數(shù)據(jù)條件下如果一張表很小,，執(zhí)行join操作最優(yōu)的選擇無(wú)疑是broadcast hash join,，效率最高。但是一旦小表數(shù)據(jù)量增大,，廣播所需內(nèi)存,、帶寬等資源必然就會(huì)太大，broadcast hash join就不再是最優(yōu)方案,。此時(shí)可以按照join key進(jìn)行分區(qū),，根據(jù)key相同必然分區(qū)相同的原理，就可以將大表join分而治之,，劃分為很多小表的join,，充分利用集群資源并行化。如下圖所示,，shuffle hash join也可以分為兩步：

1. shuffle階段：分別將兩個(gè)表按照join key進(jìn)行分區(qū),，將相同join key的記錄重分布到同一節(jié)點(diǎn)，兩張表的數(shù)據(jù)會(huì)被重分布到集群中所有節(jié)點(diǎn),。這個(gè)過(guò)程稱(chēng)為shuffle

2. hash join階段：每個(gè)分區(qū)節(jié)點(diǎn)上的數(shù)據(jù)單獨(dú)執(zhí)行單機(jī)hash join算法,。

看到這里,，可以初步總結(jié)出來(lái)如果兩張小表join可以直接使用單機(jī)版hash join；如果一張大表join一張極小表,，可以選擇broadcast hash join算法,；而如果是一張大表join一張小表，則可以選擇shuffle hash join算法,；那如果是兩張大表進(jìn)行join呢,？

Sort-Merge Join

SparkSQL對(duì)兩張大表join采用了全新的算法－sort-merge join，如下圖所示,，整個(gè)過(guò)程分為三個(gè)步驟：

1. shuffle階段：將兩張大表根據(jù)join key進(jìn)行重新分區(qū),，兩張表數(shù)據(jù)會(huì)分布到整個(gè)集群，以便分布式并行處理

2. sort階段：對(duì)單個(gè)分區(qū)節(jié)點(diǎn)的兩表數(shù)據(jù),，分別進(jìn)行排序

3. merge階段：對(duì)排好序的兩張分區(qū)表數(shù)據(jù)執(zhí)行join操作,。join操作很簡(jiǎn)單，分別遍歷兩個(gè)有序序列,，碰到相同join key就merge輸出,，否則取更小一邊，見(jiàn)下圖示意：

仔細(xì)分析的話會(huì)發(fā)現(xiàn),，sort-merge join的代價(jià)并不比shuffle hash join小,，反而是多了很多。那為什么SparkSQL還會(huì)在兩張大表的場(chǎng)景下選擇使用sort-merge join算法呢,？這和Spark的shuffle實(shí)現(xiàn)有關(guān),，目前spark的shuffle實(shí)現(xiàn)都適用sort-based shuffle算法，因此在經(jīng)過(guò)shuffle之后partition數(shù)據(jù)都是按照key排序的,。因此理論上可以認(rèn)為數(shù)據(jù)經(jīng)過(guò)shuffle之后是不需要sort的,，可以直接merge。

經(jīng)過(guò)上文的分析,，可以明確每種Join算法都有自己的適用場(chǎng)景,，數(shù)據(jù)倉(cāng)庫(kù)設(shè)計(jì)時(shí)最好避免大表與大表的join查詢，SparkSQL也可以根據(jù)內(nèi)存資源,、帶寬資源適量將參數(shù)spark.sql.autoBroadcastJoinThreshold調(diào)大,，讓更多join實(shí)際執(zhí)行為broadcast hash join。

總結(jié)

Join操作是傳統(tǒng)數(shù)據(jù)庫(kù)中的一個(gè)高級(jí)特性,，尤其對(duì)于當(dāng)前MySQL數(shù)據(jù)庫(kù)更是如此,，原因很簡(jiǎn)單，MySQL對(duì)Join的支持目前還比較有限,，只支持Nested-Loop Join算法,，因此在OLAP場(chǎng)景下MySQL是很難吃的消的，不要去用MySQL去跑任何OLAP業(yè)務(wù)，結(jié)果真的很難看,。不過(guò)好消息是MySQL在新版本要開(kāi)始支持Hash Join了,，這樣也許在將來(lái)也可以用MySQL來(lái)處理一些小規(guī)模的OLAP業(yè)務(wù)。

和MySQL相比,，PostgreSQL,、SQLServer、Oracle等這些數(shù)據(jù)庫(kù)對(duì)Join支持更加全面一些,，都支持Hash Join算法,。由PostgreSQL作為內(nèi)核構(gòu)建的分布式系統(tǒng)Greenplum更是在數(shù)據(jù)倉(cāng)庫(kù)中占有一席之地，這和PostgreSQL對(duì)Join算法的支持其實(shí)有很大關(guān)系,。

總體而言,，傳統(tǒng)數(shù)據(jù)庫(kù)單機(jī)模式做Join的場(chǎng)景畢竟有限，也建議盡量減少使用Join,。然而大數(shù)據(jù)領(lǐng)域就完全不同，Join是標(biāo)配,，OLAP業(yè)務(wù)根本無(wú)法離開(kāi)表與表之間的關(guān)聯(lián),，對(duì)Join的支持成熟度一定程度上決定了系統(tǒng)的性能，夸張點(diǎn)說(shuō),，’得Join者得天下’,。本文只是試圖帶大家真正走進(jìn)Join的世界，了解常用的幾種Join算法以及各自的適用場(chǎng)景,。后面兩篇文章還會(huì)涉及Join的方方面面,，敬請(qǐng)期待！