面向Flink的多表連接計算性能優(yōu)化算法

李 旺 雙 鍇

(北京郵電大學網絡技術研究院 北京 100876)

0 引 言

社交媒體、移動設備及傳感器以前所未有的速度持續(xù)產生著海量數(shù)據，這些數(shù)據經過簡單的預處理之后被存儲到分布式數(shù)據倉庫中，用于后期的計算、分析與挖掘。由于數(shù)據量巨大，需要采用分布式計算架構對計算進行拆分后分發(fā)到成百甚至上千臺機器上并行執(zhí)行。Flink的出現(xiàn)正好解決了大規(guī)模數(shù)據計算問題，相比于MapReduce[1]框架，F(xiàn)link[2]具有流批一體的數(shù)據處理語義[3]、基于線程的計算模型和中間結果無須寫入磁盤等優(yōu)點。Flink針對Table API[4]提供了執(zhí)行計劃優(yōu)化模塊，該模塊對作業(yè)執(zhí)行計劃優(yōu)化后生成相應的物理執(zhí)行計劃，然后提交到集群運行。該模塊提供了靈活的拓展接口，可以對Flink作業(yè)進行自定義優(yōu)化。

在Flink分布式計算架構下，執(zhí)行多表連接(multi-table join)操作時應考慮以下兩個方面。

① 由于Flink提供了基于線程的輕量級計算模型，在集群中可以提供更高的計算并行度，而用戶編寫的程序在執(zhí)行多表連接時并不會考慮到表的大小及關聯(lián)性等特性。因此，本文需要用算法來優(yōu)化多表連接的并行度，從而提升作業(yè)的整體性能。

② 在連接過程中，需要進行大量的數(shù)據shuffle操作以完成連接計算，導致過高的網絡IO代價。因此，本文需要設計一個算法在并行執(zhí)行多表連接時盡量減少需要進行shuffle操作的數(shù)據量。

現(xiàn)有的多表連接的優(yōu)化研究主要圍繞MapReduce計算框架展開，優(yōu)化措施主要包括執(zhí)行計劃和執(zhí)行框架兩方面的優(yōu)化。由于Flink的性能優(yōu)化和編程模型的差異，已有算法不能充分利用Flink集群的性能優(yōu)勢。本文提出了一個適用于Flink的Multi Bushy Tree算法，用于提高多表連接的并行度。在多表連接過程中，已有算法主要致力于尋找最優(yōu)的連接順序以減小中間結果數(shù)據的大小，而忽略了利用集群的分布式特點盡可能將多表連接計算并行化。本文充分考慮了表之間關聯(lián)鍵的依賴關系，尋找局部星型連接，將不存在依賴關系的連接計算和星型連接中事實表和維表的連接計算并行化，從而縮短多表連接作業(yè)的執(zhí)行時間。

為了進一步提高多表連接的速度，本文在Multi Bushy Tree算法的基礎上提出了Semi Join算法。該算法在執(zhí)行星型連接中事實表和維表的連接計算時，只對事實表中的關聯(lián)鍵字段和維表數(shù)據執(zhí)行shuffle操作，連接得到的各中間結果表以事實表中的主鍵字段作為關聯(lián)條件，執(zhí)行連接計算。由于Flink提供的執(zhí)行計劃優(yōu)化功能[5]，多個中間結果表的連接操作可以chain在一個節(jié)點執(zhí)行。整個連接過程需要shuffle操作的數(shù)據量大大減少，可以有效減小網絡IO代價，提高星型連接的速度。

1 相關工作

對于執(zhí)行計劃的優(yōu)化，文獻[6]的處理就是將多表連接看作是鏈式連接問題，通過將n張表的連接操作拆分成為n-1個2元連接，每個2元連接使用一個MapReduce作業(yè)完成，各MapReduce作業(yè)之間是串行執(zhí)行的。當n較大時會導致較高的作業(yè)時間復雜度和中間結果的存儲代價，不能充分利用分布式作業(yè)的優(yōu)勢。文獻[7]利用改進的蟻群算法尋找連接樹的較優(yōu)解，在一定程度上避免了局部最優(yōu)解，同時縮短了搜索時間。文獻[8]提出的多表連接算法可以在一個MapReduce作業(yè)中完成所有的連接運算，但是隨著參與連接的表增加，這種算法通過網絡IO進行shuffle操作的數(shù)據量將急劇增加，從而導致較高的網絡IO代價。文獻[9]提出的SmartJoin算法雖然可以使用更少的MapReduce作業(yè)數(shù)來實現(xiàn)鏈式多表連接不能完成的大量表之間的連接操作，但是該算法對表的限制太多，它要求參與連接的表必須包括兩個大表，其余小表必須存儲在執(zhí)行Reduce任務的節(jié)點上，參與連接的兩個大表還必須存在關聯(lián)鍵。文獻[10]將多表連接規(guī)模劃分為小中大三種類型，對于不同類型采用不同的連接順序優(yōu)化方法，并引入線性動態(tài)規(guī)劃優(yōu)化算法的時間復雜度。上述所有算法都是通過優(yōu)化多表連接計算的連接順序來提高作業(yè)的運行效率，沒有考慮將無依賴關系的連接計算并行化，無法充分利用Flink集群的節(jié)點資源。

對于執(zhí)行框架的優(yōu)化，文獻[11]基于MapReduce框架提出在Map-Reduce兩階段之后新增一個Merge階段，在新增的Merge階段執(zhí)行表連接計算，從而節(jié)省了一次MapReduce作業(yè)。文獻[12]通過修改MapReduce框架，支持數(shù)據在各節(jié)點之間進行管道式傳輸，支持在線聚集和持續(xù)查詢，但是框架修改之后，增加了失敗恢復(failover)實現(xiàn)的復雜度，同時對表連接的性能增益有限。文獻[13]將傳統(tǒng)的Sort-Merge Join算法應用到大規(guī)模分布式系統(tǒng)中，突破了單機的內存限制，同時避免了在reduce節(jié)點執(zhí)行兩表連接操作時進行笛卡爾積的計算。文獻[14]通過計算得到的關聯(lián)鍵的布爾分布矩陣，在Map節(jié)點利用矩陣對表數(shù)據進行過濾，可以有效減小網絡IO代價，但是該算法對外連接增益有限而且增加了布爾分布矩陣的計算時間。文獻[15]利用廣播變量，在執(zhí)行兩表連接時，將小表存放到廣播變量中，讀取大表數(shù)據的節(jié)點會從廣播變量中讀取小表中的全部數(shù)據，進而執(zhí)行連接計算，但該方式要求小表中的數(shù)據量足夠少，否則對小表采用復制廣播的方式依然會對網絡IO產生極大的壓力。Flink基于DataFlow的編程模型[16]與MapReduce實現(xiàn)存在較大差異，無法將已有執(zhí)行框架優(yōu)化算法直接應用到Flink上。

2 面向Flink的多表連接優(yōu)化算法

論文主要的設計與優(yōu)化目標是在使用Flink計算引擎執(zhí)行多表連接作業(yè)時，提高作業(yè)的運行效率。鑒于此，本文從兩個方面進行考慮。

1) 在分布式集群中，計算的并行度對作業(yè)的運行效率有很大影響，為提高多表連接計算的并行度，所提出Multi Bushy Tree算法，可以有效地將不存在依賴關系的連接計算并行執(zhí)行。

2) 對于星型連接，本文提出Semi Join算法，通過拆分事實表的關聯(lián)鍵的方式減少在連接計算時需要shuffle的數(shù)據量，以此來減小星型連接的網絡IO代價。

2.1 連接樹模型

如圖1所示，多表連接可以用一個查詢圖G=(V,E)來表示，其中：V是圖中所有節(jié)點的集合，每個節(jié)點代表一個參與連接的數(shù)據表；E是圖中所有邊的集合，如果兩個節(jié)點之間存在連接邊表示這兩張數(shù)據表存在關聯(lián)關系。圖1(a)的查詢圖包含6個關聯(lián)關系，為6元連接。同理，圖1(b)的查詢圖為8元連接。

(a) Left-deep Tree(b) Right-deep Tree

(a) 6-table join (b) 8-table join圖1 多表連接查詢示例

當表的數(shù)量較少時，可以通過窮舉的方式得到一個最優(yōu)的連接順序。但是當參與連接的表的數(shù)量超過一定大小時，該問題則不能得到有效的解決。有研究表明多表連接執(zhí)行計劃的最優(yōu)解確定是一個NP-hard問題[18]，傳統(tǒng)的做法通常采用連接樹模型限定解的一個子空間，并設計算法從生成的連接樹中找到一個較優(yōu)解從而確定連接的順序。連接樹可以描述不同的連接方案但最終都會得到相同的連接結果。如圖2所示，主流的連接樹模型有Left-deep Tree、Right-deep Tree、Zigzag Tree和Bushy Tree四種。

(c) Zigzag Tree (d) Zigzag Tree圖2 四種不同的連接樹

本文首先對連接樹中涉及到的概念進行說明：用戶輸入的連接條件涉及到的數(shù)據表稱為基本表，兩個表進行連接計算后得到的中間結果稱為中間表。連接樹是一棵二叉樹，葉子節(jié)點永遠是基本表，內部節(jié)點永遠是中間表，邊表示數(shù)據的流向。對于Left-deep Tree、 Right-deep Tree和Zigzag Tree，其內部節(jié)點至少有一個葉子節(jié)點作為子節(jié)點，表示每個中間表都至少由一個基本表進行連接計算得到。對于Bushy Tree，有些內部節(jié)點的子節(jié)點是不包含葉子節(jié)點的。實際上其他三種連接樹都是Bushy Tree的一種特殊形式。

2.2 多元連接樹算法

Flink提供的基于線程的輕量級計算模型，使得分布式計算作業(yè)能以更高的并行度運行。在多表連接方向，目前已有的算法主要是通過減小中間表大小的方式提高作業(yè)的運行效率，且都是針對MapReduce作業(yè)，無法充分發(fā)揮Flink的并行計算優(yōu)勢，因此并不完全適用于Flink計算作業(yè)。本文提出的算法充分考慮了Flink計算引擎的優(yōu)點，盡量提高多表連接計算的運行效率。基于構建得到的較優(yōu)Bushy Tree，分析表之間的依賴關系，盡可能提高多表連接計算的并行度。多元連接樹(Multi Bushy Tree)算法詳細步驟見算法1。

算法1Multi Bushy Tree算法

輸入:JP“連接表的集合”。

輸出:PT“多元連接樹”。

1.BEGIN

2.bushyTree←BuildBushyTree(JP);

3.//將JP根據表之間的關聯(lián)關系構建二元Bushy Tree

4.FOR EACHnodeINbushyTreeDO

5.IFshouldMerge(node) THEN

6.node←mergeNode(node);

7.END FOR

8.//如果節(jié)點的所有子節(jié)點(C1,C2,…,Cn)都是基本表且Cn

//基本表的關聯(lián)鍵來自于C1，合并所有子節(jié)點并替換該節(jié)點

9.FOR EACHnodeINbushyTreeDO

10.IFshouldExecute(node) THEN

11.IFisStarJoin(node) THEN

12.parallelExecuteStarJoin(node);

13.ELSE

14.ParallelExecuteNormalJoin(node);

15.ENDFOR

16.RETURNPT;

17.END

圖3 二元濃密連接樹示意圖

② 尋找局部星型連接。對于步驟①構建得到的二元濃密連接樹，遍歷樹中的每一個節(jié)點，如果該節(jié)點的右孩子節(jié)點表示的基本表中的關聯(lián)鍵均來自于最左孩子節(jié)點表示的基本表或者中間表，則用所有的孩子節(jié)點替換該節(jié)點，將二叉樹變?yōu)槎嗖鏄?。例如，對于步驟①中的查詢示例，在分支一中，連接表C時，連接條件為“A.key2=C.key”，表C中的關聯(lián)鍵key同樣來自于表A中的key2列，滿足合并的要求，因此可以使用A、B兩個節(jié)點代替節(jié)點M1，更新后的連接樹見圖4(a)。針對分支二可以做相同處理，得到的連接樹如圖4(b)所示。而表A連接表D時，由于表D和表A的連接是笛卡爾積操作，因此不滿足合并條件。對整個二元連接樹處理之后，得到的最終的多元濃密連接樹(Multi Bushy Tree)，如圖4(b)所示。對于優(yōu)化得到的多元濃密連接樹，如果某個節(jié)點包含n(n>2)個子節(jié)點，表明該中間結果表是表(T1,T2,…,Tn)通過星型連接運算得到的，其中T1在星型連接中作為事實表，T2-Tn在星型連接中作為維表。如果某個節(jié)點只有兩個子節(jié)點，表明該中間結果表是通過鏈式連接或者笛卡爾積運算得到。

(a) (b)圖4 多元濃密連接樹

Multi Bushy Tree算法充分考慮表之間的關聯(lián)性，將不存在依賴關系的連接計算并行化，提升多表連接作業(yè)的整體性能。此外，Multi Bushy Tree算法在構建的較優(yōu)連接樹基礎上，充分挖掘局部星型連接，并將星型連接中事實表和各個維表之間的連接計算并行化，極大地提高了多表連接計算的并行度。然而在星型連接中，仍然需要將事實表中的數(shù)據通過網絡IO進行多次shuffle以完成和各維表的連接計算，這對星型連接的計算性能仍會有較大影響。因此，針對星型連接提出了一個全新的連接算法減小網絡IO代價。

2.3 關聯(lián)鍵拆分連接算法

這一部分中，針對星型連接，本文提出基于Flink計算引擎可以有效減少需要進行shuffle的數(shù)據量的算法。由于在星型連接中，事實表通常作為表連接計算中的左表，此時如果事實表和多張數(shù)據表進行連接計算時的關聯(lián)鍵相同，F(xiàn)link提供的執(zhí)行計劃優(yōu)化器就會把這些連接計算合并在一個節(jié)點運行。利用這一特性，本文提出了關聯(lián)鍵拆分連接(Semi Join)算法，利用該算法對星型連接優(yōu)化后，只需要對事實表、維表和中間結果表各執(zhí)行一次shuffle操作，可以進一步提高星型連接的執(zhí)行效率，減小網絡IO代價。詳細算法步驟見算法2。

算法2Semi Join算法

輸入:JP“星型連接樹”。

輸出:PT“星型連接結果”。

1.BEGIN

2.FORi=0 TOnumOf(dimTables) DO

3.joinCols←getJoinCols(dimTable);

4.localFactTables[i]←select(factTable,joinCols);

5.ENDFOR

6.//根據各維表的關聯(lián)鍵從事實表中選擇外鍵字段和主鍵

//字段生成局部事實表集合

7.FORi=0 TOnumOf(factTempTables) DO

8.dimResTables[i]←join(localFactTables[i],dimTables[i]);

9.ENDFOR

10.//對局部事實表和維表執(zhí)行連接計算，得到各維表的連接

//結果表集合

11.FORi=0 TOnumOf(dimResTables) DO

12.factTable←join(factTable,dimResTables[i]);

13.ENDFOR

14.//各維表的連接結果表和事實表以事實表的主鍵字段作為

//連接條件進行連接計算，得到最終結果表

15.RETURNfactTable;

16.END

① 生成局部事實表。通過解析Flink執(zhí)行計劃，得到事實表中和維表關聯(lián)的外鍵字段。使用Flink提供的“select”函數(shù)選擇解析出的外鍵字段和事實表的主鍵字段作為局部事實表。例如，查詢命令“SELECT*FROM A,B WHERE A.fKey=B.key”，事實表A使用fKey字段和維表B進行關聯(lián)，因此生成的局部事實表T中包含兩個字段：fKey和key(事實表的主鍵)。由于Flink提供的執(zhí)行計劃優(yōu)化功能，select計算和前面的計算函數(shù)會合并在一個節(jié)點中執(zhí)行，因此不會產生數(shù)據shuffle。

② 生成各維表的連接結果表。在步驟①中為每個維表生成了連接需要的局部事實表，由于各局部事實表只包含了事實表中的用于連接指定維表的外鍵字段和事實表的主鍵字段，因此每個局部事實表的數(shù)據量都比較小。將生成的局部事實表和維表按照關聯(lián)字段進行hash shuffle，關聯(lián)字段相等的數(shù)據會通過網絡IO發(fā)送到相同節(jié)點執(zhí)行連接計算。由于局部事實表和維表數(shù)據量都比較小，采用hash shuffle的方式可以避免將局部事實表進行復制廣播的開銷，有效減少了需要進行shuffle的數(shù)據量。以步驟①的查詢?yōu)槔?，局部事實表T(key,fKey)和維表D(key,value1,value2)，執(zhí)行連接計算時，表T計算fKey字段的hash值，對并行度取模之后發(fā)送到指定節(jié)點，表D對key字段hash取模后發(fā)送到指定節(jié)點，表T中fKey和表D中key字段相同的數(shù)據會發(fā)送到同一個節(jié)點，這種連接方式保證表中的一條數(shù)據只會進行一次網絡IO。

③ 生成最終查詢結果表。通過步驟(2)得到的各維表的連接結果表中除了包含各維表中的查詢字段還包含了事實表中的主鍵字段，事實表和各連接結果表通過事實表中的主鍵執(zhí)行鏈式連接計算從而得到最終的查詢結果。由于事實表和各連接結果表的鏈式連接計算中所有連接的關聯(lián)鍵都相同，F(xiàn)link提供的執(zhí)行計劃優(yōu)化功能會將這些連接計算chain在一個節(jié)點中執(zhí)行，所以在本次鏈式連接中事實表和各連接結果表只需要執(zhí)行一次數(shù)據shuffle操作。

圖5給出了基于Flink的星型連接優(yōu)化算法Semi Join的執(zhí)行流程。

圖5 星型連接示意圖

利用Semi Join算法優(yōu)化之后的星型連接計算，步驟①、步驟②和步驟③是串行的，而步驟②中局部事實表和各維表之間的連接可以并行計算，因此星型連接的時間代價計算式表示為：

CostlocalFact=Max(C1,C2,…,Cn)

CostdimJoin=Max(R1,R2,…,Rn)

(1)

Cost=CostlocalFact+CostdimJoin+CostfinalJoin

式中：C1,C2,…,Cn表示從事實表中查詢得到與各維表進行連接的局部事實表的時間代價，由于SELECT計算是并行執(zhí)行的，所以該階段時間代價取決于最慢的SELECT計算。R1,R2,…,Rn表示各局部事實表和各維表連接計算的時間代價，各連接計算在集群中并行執(zhí)行，該階段的時間代價同樣取決于最慢的連接計算的時間代價。最后整個星型連接計算的時間代價為三個串行計算的時間代價之和。

從圖5可以看出，基于Flink的星型連接優(yōu)化算法Semi Join主要包含兩種類型的連接操作：局部事實表-維表連接和事實表-結果表連接。出于對計算性能和存儲性能考慮，算法中的連接計算采用Hash Join的方式，每次連接都會進行數(shù)據shuffle操作，因此網絡IO代價計算式表示為：

(2)

Cost=CostdimJoin+CostfinalJoin

式中：L[i]表示第i個局部事實表的大?。籇[i]表示第i個維表的大??；R[i]表示第i個連接結果表的大?。籉表示事實表的大小。生成連接結果表時，需要對參與連接的局部事實表和維表進行shuffle操作。生成最終結果表時，需要對事實表和各連接結果表進行shuffle操作。整個星型連接需要shuffle操作的總數(shù)據量為兩個階段shuffle操作數(shù)據量之和。

3 實驗與結果分析

本文將Multi Bushy Tree+Semi Join與其他兩種多表連接算法進行比較。(1) Left-deep Tree連接。Flink提供了基于Left-deep Tree的多表連接方式，連接的順序取決于用戶輸入。(2) Bushy Tree連接。通過Bushy Tree的方式構建連接樹，F(xiàn)link可以將不存在依賴關系的兩表連接計算并行化。本文實現(xiàn)了這3種連接算法，并從計算并行度、作業(yè)運行時間和網絡IO三個方面進行評估。

本文搭建了具有20個節(jié)點的集群，其中一個節(jié)點被用作Master和ResourceManager，負責任務調度和資源管理，另外的19個節(jié)點被用作Worker和TaskManager，所有的計算任務都在這些節(jié)點上運行。每臺機器的硬件配置為4核2.4 GHz的CPU、8 GB內存、40 GB機械硬盤。集群中的每個節(jié)點均安裝了CentOS 7 64位操作系統(tǒng)，使用的Flink版本為原生的Flink 1.9.0，底層使用YARN 2.9.2作為分布式調度系統(tǒng)。

實驗采用的數(shù)據集為TPC-H提供的Dbgen工具所生成的模擬數(shù)據集。為了評價三種算法在不同的多表連接方式下的性能差異，實驗生成了A、B、C和D四種數(shù)據集，數(shù)據集格式和大小如表1所示，四種數(shù)據集基本可以覆蓋多表連接的各種情況。

表1 數(shù)據集格式

3.1 計算并行度分析

在分布式集群中，提高多表連接計算的并行度可以有效地縮短計算作業(yè)的運行時間。由于多表連接計算根據依賴關系可以劃分為不同的計算階段，每個計算階段的并行度并不相等，因此采用各階段的平均并行度作為衡量標準，設置每個兩表連接計算的并行度為4。表2展示了不同連接算法在執(zhí)行多表連接計算時的并行度。

表2 多表連接并行度

對于數(shù)據集A，由于表之間不存在局部星型連接，Multi Bushy Tree算法無法在Bushy Tree的基礎上做進一步優(yōu)化。在數(shù)據集B中，Multi Bushy Tree算法可以基于Bushy Tree算法進一步并行執(zhí)行局部星型連接中事實表和維表的連接計算，所以并行度進一步提高。在數(shù)據集C中，雖然存在兩個局部星型連接，但是兩者存在關聯(lián)依賴關系，無法進一步并行化兩個星型連接計算，結果表明Multi Bushy Tree算法的平均并行度仍然高于Bushy Tree算法優(yōu)化后的并行度。在數(shù)據集D中，由于Multi Bushy Tree算法可以并行執(zhí)行兩個局部星型連接和星型連接內部事實表和維表的連接計算，平均并行度進一步提高。由于Left-deep Tree算法只能串行執(zhí)行多個兩表連接計算，因此在四個數(shù)據集中，Multi Bushy Tree和Bushy Tree算法的并行度均高于Left-deep Tree算法。

實驗表明，Multi Bushy Tree算法在多表連接存在局部星型連接的情況下，可以有效提高連接計算的并行度，并且隨著表數(shù)量的增加，并行度增加更明顯。對于不存在局部星型連接的情況，利用二元濃密連接樹仍可并行執(zhí)行不存在依賴關系的兩表連接計算。

3.2 作業(yè)運行時間分析

圖6展示了對于不同數(shù)據集，每種多表連接算法整體作業(yè)的執(zhí)行時間。由于在Left-deep Tree算法中，多表連接計算只能拆分為n-1個串行執(zhí)行的兩表連接計算，且有可能較早產生笛卡爾積計算，整體作業(yè)的執(zhí)行時間比其他兩種算法高很多。在剩下的兩種多表連接算法中，Multi Bushy Tree算法比Bushy Tree算法的作業(yè)運行時間低很多。主要是因為雖然Multi Bushy Tree+Semi Join花費了不少時間用于尋找局部星型連接和生成局部事實表，但實驗結果表明提高多表連接計算的并行度可以極大降低連接作業(yè)的運行時間。從整體上看，Multi Bushy Tree通過尋找局部星型連接并并行化星型連接中事實表和維表之間的連接計算，從而降低作業(yè)的運行時間，而Semi Join算法減少了星型連接過程中網絡IO時間。

圖6 不同多表連接算法在不同數(shù)據集下的作業(yè)運行時間

實驗表明，在四種類型的數(shù)據集中，Multi Bushy Tree算法均表現(xiàn)最好。相比于其他兩種連接算法，在數(shù)據集A上優(yōu)化效果最差，作業(yè)運行時間縮短0%和20.80%，在數(shù)據集D上優(yōu)化效果最好，作業(yè)運行時間縮短14.98%和44.78%。在所有數(shù)據集上，Multi Bushy Tree+Semi Join算法表現(xiàn)最好。

3.3 網絡IO代價分析

在星型連接中，Semi Join算法通過選擇與各維表進行連接的關聯(lián)鍵字段和事實表的主鍵字段得到局部事實表，使用局部事實表和維表進行連接計算得到各維表的連接結果表，最后將多個連接結果表通過事實表的主鍵字段進行連接計算得到最終查詢結果表的方式，避免了對事實表進行多次shuffle的操作，能夠極大減少通過網絡IO進行shuffle操作的數(shù)據量。實驗使用數(shù)據生成工具Dbgen生成了三個只包含星型連接的數(shù)據集Q1、Q2和Q3，除了各數(shù)據集中均包含一張事實表外，Q1中包含3張維表，Q2中包含5張維表，Q3中包含10張維表。在三個數(shù)據集上分別比較Semi Join算法和Flink提供的級聯(lián)連接算法為完成連接所產生的網絡IO代價。實驗結果表明，Semi Join算法可以有效減少進行shuffle的數(shù)據量。

根據圖7中的數(shù)據可知，在星型連接中，維表的數(shù)量越多，Semi Join算法對于網絡IO的優(yōu)化效果越明顯。在三個數(shù)據集中，網絡IO的數(shù)據量分別減少了65.4%、77.6%和89.8%。主要有以下兩個原因。

圖7 Semi Join和Chain Join產生的網絡IO數(shù)據量

(1) Semi Join算法通過生成局部事實表的方式，使得在和各維表連接時，將對整個事實表shuffle操作的需求轉化為對各局部事實表的shuffle操作。

(2) 生成最終查詢結果表時，由于各連接結果表都是通過事實表的主鍵進行關聯(lián)的，因此所有的連接計算都可以chain在一個節(jié)點運行，只需要對事實表進行一次shuffle操作。

綜上所述，雖然Semi Join算法增加了生成局部事實表的時間開銷，但是可以顯著減小需要通過網絡IO進行shuffle操作的數(shù)據量，可以有效縮短整體作業(yè)的運行時間，且算法對維表的數(shù)量不敏感。所以Semi Join算法在減少網絡IO的同時縮短了星型連接作業(yè)的運行時間。

4 結 語

基于Flink分布式計算引擎，本文為優(yōu)化多表連接的計算速度提出優(yōu)化計算并行度的Multi Bushy Tree算法和Semi Join算法。Multi Bushy Tree算法在二元濃密樹的基礎上，充分考慮了表之間的關聯(lián)性，在連接樹中尋找局部星型連接，通過并行化不存在依賴關系的連接計算和星型連接中事實表和各維表的連接計算縮短多表連接作業(yè)的運行時間。此外，Semi Join算法利用Flink提供的執(zhí)行計劃優(yōu)化功能，大量減少星型連接中需要通過網絡IO進行shuffle操作的數(shù)據量。實驗結果表明，與其他連接方法相比，Multi Bushy Tree+Semi Join算法可以大幅度提高多表連接時的計算并行度，極大縮短了多表連接作業(yè)的運行時間，有效減小網絡IO代價。

由于目前在分布式計算領域有多種計算框架，例如Spark、Beam等，都可以提供大規(guī)模分布式集群計算。在未來的工作中，我們會將Multi Bushy Tree+Semi Join的多表連接優(yōu)化方案應用到更多的計算框架中，并進一步提升算法的性能。