宗楓博，趙宇海+，王國仁，季航旭

1.東北大學 計算機科學與工程學院，沈陽110169

2.北京理工大學 計算機學院，北京100081

隨著分布式技術的快速發(fā)展，人們對大數據的處理能力有了飛躍式的提高，眾多優(yōu)秀的大數據引擎也應運而生。大數據處理引擎有著許多與傳統數據庫相似的計算操作，同樣也有許多近似的優(yōu)化方法。同時大數據處理引擎也與傳統數據庫有很多質的區(qū)別，大數據處理引擎不再依賴于企業(yè)級服務器而是可以通過部署在成百上千臺PC 上協同工作，網絡對運行效率的影響相比傳統數據庫而言更加顯著。

作為大數據處理引擎，當連接操作涉及到多個表時，連接排序對降低執(zhí)行成本有相當大的效果。連接操作符通過兩個表的公共屬性將它們連接起來，優(yōu)化器需要找到一個成本最低或幾乎接近的執(zhí)行計劃。連接順序的優(yōu)化問題通常被認為是一個NP-Complete問題，執(zhí)行計劃的個數隨著連接表的數量和數據源之間的連接關系數量增加而呈階乘級增加，目前為止針對連接順序的優(yōu)化已經提出了許多方法，包括確定性算法、啟發(fā)式算法、隨機算法、混合算法等。但是與傳統數據庫不同的是，分布式大數據處理引擎的節(jié)點要遠遠多于傳統分布式數據庫，在MapReduce 框架下各節(jié)點間I/O 對網絡傳輸造成的影響會隨著中間數據量的增長而成倍放大，同時I/O 對執(zhí)行效率的影響也被大幅放大。而在傳統的連接優(yōu)化算法中，通常只考慮連接順序優(yōu)化，通過減少全局的連接計算量來減少執(zhí)行時間，而并不考慮同時優(yōu)化每次連接的投影或通過減少全局的投影列數來減少I/O 代價，如圖1 連接樹中包括三個表、、，傳統方法中的優(yōu)化主要將表作為最小優(yōu)化粒度考慮，比如通過表的行數直接估算比較??和??哪個為更優(yōu)連接策略，但若將表細分為列來考慮，表下的[0,1]表示表包含兩個不同屬性的列，相同的數字表示列的屬性相同。在保證最終連接結果不變的條件下，可將原連接樹改變連接順序為圖2 所示連接樹，中間表由(σ(?))變?yōu)?σ(?))，由于投影到中間連接表的列由變?yōu)榱?，減少了一列，則可能進而減少了I/O 代價。為了盡可能減少全局中間連接表體積進而降低I/O 代價，本研究設計了一個連接索引結構，結合動態(tài)規(guī)劃算法在優(yōu)化連接順序的同時對每個連接表的投影關系進行優(yōu)化，同時針對連接計算代價和I/O 代價進行了優(yōu)化，有效地減少了多連接任務的運算時間和中間數據的傳輸量。

圖1 原連接樹Fig.1 Origin join tree

圖2 改變連接順序Fig.2 Modify join order

本文的主要貢獻如下：

（1）設計了一種描述連接表中列關系的索引結構，此索引結構可以快速找到兩個表的連接關系及每一列是否冗余列。

（2）基于此索引，結合動態(tài)規(guī)劃算法在優(yōu)化連接順序的同時刪除每次連接的冗余列，盡可能地減少所有中間連接表的體積。

（3）通過基于Flink 系統的實驗對本優(yōu)化系統進行了驗證，實驗結果顯示本優(yōu)化系統有效地減少了多連接任務的執(zhí)行時間和中間數據的傳輸量。

1 相關工作

確定性算法通過完全遍歷或在解空間中應用一些啟發(fā)式剪枝方法，來執(zhí)行解空間的確定性搜索，如IBM 的System R 中使用的動態(tài)規(guī)劃算法，這個算法被幾乎現有的所有商用RDBMS 系統使用，該算法通過動態(tài)剪枝解空間來進行完全遍歷，通過迭代已有的連接關系并盡可能剪枝一些次優(yōu)解來構造所有備選連接樹，從而保證了動態(tài)規(guī)劃算法在解空間中找到最優(yōu)解。隨機化算法通過預定義的變換規(guī)則在解空間內隨機移動，不斷尋找代價更低的點，當點不再移動的次數達到預定義的迭代次數閾值時，則認為當前的解為較優(yōu)的解。如模擬退火算法（simulated annealing algorithm），它允許當前點向代價更高的點移動，從而降低了算法陷入局部最優(yōu)解的概率，文獻[6]中提出了一個算法參數來定義在給定時間點繼續(xù)搜索的可能性，并和當前點與目標點的代價比例共同決定點移動的概率。

基因算法通過模擬生物的進化過程來尋找最優(yōu)解，通過將一個初始群體（解集合）中的個體（解）隨機交叉（crossover）和變異（mutation）產生下一代群體，將適應度最高（通過代價函數定義）的部分個體保留，并重復交叉和變異的過程產生下一代群體，當迭代次數達到預定義的閾值或者群體適應度已接近飽和，繼續(xù)進化無法再顯著提高適應度（通過閾值定義）時停止迭代過程。如PostgreSQL 中的Genetic Query Optimizer，它只考慮了左深樹的情況，實現了精英選擇算子（elitist selection operator），對于群體中的最優(yōu)解，只進行交叉而不進行變異，防止群體中的最優(yōu)個體在迭代到下一代時發(fā)生丟失，從而導致算法難以收斂到全局最優(yōu)解。

混合算法混合了以上兩種及以上的優(yōu)化策略，比如巡回模擬退火算法（toured simulated annealing），該算法首先通過確定性算法確定幾個初始點，然后對每個初始點進行模擬退火算法。

以上算法為主要基于傳統關系數據庫理論的多連接優(yōu)化，優(yōu)化的最小粒度為表，而在本文所提出的多表連接優(yōu)化方法中優(yōu)化的最小粒度為列。在傳統數據庫甚至一些節(jié)點數不高的分布式數據庫中，連接表的列數，或者說連接表的體積對I/O 并不會造成顯著影響，而連接表的行數，即連接表的元組數會對計算量產生很大影響，因此這些傳統的多連接優(yōu)化算法主要注重于對連接元組數進行優(yōu)化以提高計算效率。隨著MapReduce 框架在大數據處理中的廣泛應用，一個任務的計算節(jié)點可能達到成百上千個，那么節(jié)點間I/O 帶來的影響不再可以忽視。針對此問題，本文提出一種一般性的多表連接優(yōu)化方法，可以在優(yōu)化算法中同時考慮計算和I/O 的代價，并以動態(tài)規(guī)劃算法為例，在動態(tài)規(guī)劃算法的基礎上同時實現對計算和I/O 的優(yōu)化。并在實驗中對多表多列數和多表低列數的情況進行了測試，證明了本文的優(yōu)化方法不失一般性。

2 預備知識

對于批數據連接計算，Flink 系統在節(jié)點之間的傳輸有兩種策略：RR（repartition-repartition）和BF（broadcast-forward）。而節(jié)點中的本地批數據連接策略和傳輸策略是完全獨立的，在批數據傳輸完成后，每個連接實例將使用本地連接算法來連接分配的數據，本地連接策略包括兩種：SM（sort-merge-join）和HH（hybrid-Hash-join）。

2.1 Flink 中的批數據傳輸策略

如圖3 所示，RR 策略將兩個批數據源R（等分為R1 和R2）和S（等分為S1 和S2）及它們的連接屬性通過相同的分區(qū)函數發(fā)送到各個節(jié)點，每個節(jié)點被分配一個連接實例，確保有相同連接屬性的批數據被發(fā)送到相同節(jié)點，RR 策略的代價即為將兩個數據集通過網絡進行完全shuffle。

圖3 RR 傳輸策略Fig.3 RR ship strategy

如圖4 所示，BF 策略中，批數據源S 保留在本地節(jié)點不通過網絡傳輸，而每個節(jié)點的批數據源R 則被發(fā)送到其他所有節(jié)點，保證每個節(jié)點都能收到完整的批數據源R，因此BF 策略的代價僅取決于批數據源R 的大小和連接實例的數量，與批數據源S 的大小無關。顯然當批數據源R 的大小遠小于批數據源S時，系統更傾向于使用BF 策略。

圖4 BF 傳輸策略Fig.4 BF ship strategy

2.2 Flink 中的批數據連接策略

如圖5 所示，SM 的工作方式是本地節(jié)點的兩個輸入數據集按照連接屬性分別進行排序（sort），然后將排序后的數據集進行合并（merge），最后對合并后的數據集進行連接。排序過程在內存中進行，當內存資源耗盡后則將內存中已排序完成的數據寫入本地文件系統中，然后繼續(xù)對傳入的數據在內存中排序，全部數據傳入完成后在本地文件系統中會存放每個輸入流的若干不相交的有序子集，最后通過有序合并每個輸入流的有序子集，同時對合并后的兩個有序數據集進行連接。

圖5 SM 連接策略Fig.5 SM join strategy

如圖6 所示，HH 方式將兩個輸入數據集分為build 端和probe 端，并且分為兩個階段工作：probe 階段和build 階段。在build 階段中，將build 端的數據根據連接屬性插入到哈希表中，若內存資源耗盡則寫入到本地文件系統中。在probe 階段中，probe 端數據根據哈希表索引與build 端數據連接，若索引到內存中的哈希表，則直接連接并輸出，若索引到本地文件系統中的部分，則同樣也將數據寫入到本地文件系統中。在probe 端數據讀完后，若本地文件系統中存在有溢出數據，則刪除當前內存中的哈希表，通過存儲的build 端溢出數據重新構建新的哈希表，并將本地文件系統中對應的probe 端溢出數據通過新的哈希表索引并連接，遞歸地進行此流程直到所有probe端數據都被處理完成。若內存資源能夠容納完整的哈希表，則HH 方式相當于普通的In-Memory Hash Join，有著極高的效率；即使在少量數據寫入到本地文件系統的情況下，HH 仍然有著良好的性能。

圖6 HH 連接策略Fig.6 HH join strategy

3 多表連接的投影和順序優(yōu)化

首先舉例說明需要解決的問題，如圖7 所示的一段連接關系的構建中，表與表連接生成中間表，key 為列1，這時需要確定哪些列需要通過投影關系保留到中，刪除冗余列。

圖7 列保留示例Fig.7 Example of column retention

對于列0，發(fā)現還有未連接的數據源包含此列，那么此列可能在后續(xù)連接中作為key，則此次投影中保留此列。

對于列1，在作為此次連接的key 后，發(fā)現此列不再存在于還未連接的任何數據源中，也不存在于最終結果result中，則可判定此列為冗余列，此次投影不再保留，列2 同理。

對于列3，發(fā)現原連接指定的result 中包含此列，則在每次投影中都需要保留此列。

通過解決以上所提到的冗余列問題，在同時考慮到連接順序對連接效率的條件下，對原連接樹圖8使用本文提供的優(yōu)化方法，優(yōu)化后的連接樹為圖9，連接使用的數據集為TPC-H 標準測試數據集，其與Source表的對應關系如表1 所示。

圖8 原連接樹Fig.8 Origin join tree

圖9 優(yōu)化后的連接樹Fig.9 Optimized join tree

表1 Source表對應的TPC-H 表Table 1 TPC-H table corresponding to Source table

為了執(zhí)行此優(yōu)化策略，本文設計了一種關于列和表的投影關系索引，通過此索引能夠快速判斷兩個表的連接鍵值，以及連接后所需要保留的必要列。建立投影關系索引主要由兩個模塊組成：對列關系的溯源和列與連接關系索引的建立。

在得到投影關系索引后，可以基于此索引對每次連接的投影進行判斷，及時刪除冗余列減小I/O負擔，具體實現方法為3.3節(jié)基于投影關系索引的投影判斷。

為了衡量連接的優(yōu)劣，需要設計一個綜合考慮CPU 和I/O 的代價函數，基于前文2.1 節(jié)Flink 中的批數據傳輸策略可以得到I/O 的代價函數，基于前文2.2節(jié)Flink 中的批數據連接策略介紹可以得到CPU 的代價函數。通過將代價函數與動態(tài)規(guī)劃算法結合可以得到連接順序的優(yōu)化結果，具體實現方法為3.4 節(jié)基于投影關系索引的連接順序優(yōu)化。

3.1 對列關系的溯源

此算法目的在于：對于連接樹中指定一個表的一個列，能夠找到以此列為key 具有連接關系的所有Source表的集合。

如算法1 所示，給定一個表和此表中的一個列，若表為Source 表，則創(chuàng)建一個記錄節(jié)點（行3），記錄此表的id 和列在此表中的位置（行4），將記錄節(jié)點加入到中（行5）；若表為Join 表，則對投影到列的輸入表進行遞歸溯源，將記錄節(jié)點加入到中（行9），若列同時為此Join 表的key，那么同樣要對另外的輸入表進行遞歸溯源，將記錄節(jié)點加入到中（行11）。

Trace

輸入：,the table currently accessed；,the column to be traced to the current access table.

輸出：,the set of root tables found by traversing.

舉例說明，與通過列6 為key 進行連接，需要得知哪些Source表包含了列6，這些Source表均有以列6 為key 的連接關系。

如圖10 所示，對于中的列6 進行溯源，為Join表且列6為key，則列6的來源包括和，遞歸地對表溯源，找到和，則以列6 為key 具有連接關系的Source 表集合為{4,5,6}。

圖10 對列6 的溯源結果Fig.10 Result of tracing column 6

3.2 建立投影關系的索引

此算法的目的在于，對于連接過程中的列，在每次連接的投影關系策略中，需要保留可能會在之后的連接中作為key 的列和包含在Result 中的列，為此建立一個十字鏈表形式的索引結構，由Source 表索引、Key 節(jié)點索引和連接信息節(jié)點構成，每個連接信息節(jié)點中記錄所有能夠作為同一連接key 的列，通過這個索引可以快速地確認當一次連接完成后哪些列在投影操作中可以被刪除，減小對I/O 的影響。

這里說明用于建立所有key 列與Source 表的關系索引的算法2，對于所有Join 表的集合，遍歷其中的每個Join 表，對的連接key 進行溯源得到記錄節(jié)點集合（行3），創(chuàng)建一個新的連接信息節(jié)點記錄這個集合（行4），遍歷集合中的每個記錄節(jié)點，通過Source 表索引查找，若存在已有的連接信息節(jié)點包含（行7），則說明所記錄的key 和所記錄的key 是等價的，因為這些key 所源自的Source 表互相存在連接關系，所以可以將與合并（行8），并用重新指向合并后的節(jié)點（行9）；若不存在于對應Source 表的鏈表中，則說明這是一個有關于此Source 表的新的連接關系，將其加入到Source表的鏈表中（行12）。而若中的每個記錄節(jié)點都不存在于已有連接信息節(jié)點中，則說明所代表key 不與已有的任何key 存在等價關系，則將其作為一個新的連接信息節(jié)點加入到key 節(jié)點索引中（行15）。

Key columns index

輸入：,the set of JoinNode;,the linked list array of the relationship between the source table and the key information node.

輸出：,the index array of key information node.

舉例說明，在對表中的key 列3 進行溯源后，得到如圖11 所示索引。左側0～7 的縱表為Source表索引，對于一個節(jié)點，節(jié)點頭(3)表示此key 列在整體連接樹中的列編號為3，在所有的key 列中編號為2。指向節(jié)點內容1 表示此key 列在中的位置為1。指向節(jié)點內容0 表示此key 列在中的位置為0。

圖11 對Join2 表中的列3 溯源得到索引Fig.11 Index of tracing column 3 in Join2 table

如圖12 所示，在對表中的key 列6 進行溯源后，得到記錄節(jié)點集合{列1,列1}，通過Source 索引查詢，發(fā)現并不與現存的連接信息節(jié)點有交集，那么作為新的連接記錄節(jié)點{(6),列1,列1}加入到key 節(jié)點索引中。

圖12 對Join3 表中的列6 溯源得到索引Fig.12 Index of tracing column 6 in Join3 table

如圖13 所示，在對表中的key 列6 進行溯源后，得到記錄節(jié)點集合{列1,列1列0}，通過Source 索引查詢，發(fā)現與已有的節(jié)點存在交集，則將記錄集合與節(jié)點合并。

圖13 對Join4 表中的列6 溯源得到索引Fig.13 Index of tracing column 6 in Join4 table

通過此方法，最終得到如圖14 所示連接樹的索引結構。

圖14 原連接樹的連接關系索引Fig.14 Join relations of origin join tree

同樣，對于最終結果Result 表的所有列采取類似的算法得到關于投影到Result 表上所有列與Source表的關系索引。

3.3 基于投影關系索引的投影判斷

此算法的目的在于，基于已有的投影關系索引，確定每次連接時需要保留的列。

如算法3 所示，對于兩個輸入表和，按照Source 索引，將中的所有已連接的Source 表所在的記錄節(jié)點標記為finished（行7），將記錄節(jié)點所在的KeyNode 節(jié)點標記為processed 表示已被遍歷（行6）；對進行同樣的操作，若有KeyNode 被兩次遍歷，則說明此KeyNode 所代表的列同時存在于和中，即為和的連接key 列（行14～15）。若有Key-Node 中的所有列信息節(jié)點均被標記為finished，則說明此KeyNode 所代表的key 列連接Source 表已全部連接，那么此key 列在后續(xù)連接中不會再被使用（行12～13）。同時若此key 列不存在于Result 索引中，即與最終投影結果無關，那么在此次連接中不再對此列進行投影保留（行22）?；诖怂惴梢源_保每次連接投影所保留的列只包括可能會在后續(xù)連接中所用到的key 列和最終結果的來源列。

NewJoinNode

輸入：,InputNode0;,InputNode1;,linked list array of the relationship between the source table and the key information node;,index array of key information node;,index array of result information node.

輸出：,the new JoinNode obtained by connectionand.

舉例說明，如圖15 所示，連接和表，通過Source 索引查詢，發(fā)現和表同時包含列，則為和的連接key，連接結果為表，投影的key 列包括列。連接表和表，通過Source 索引查詢表和表同時包含列，則為和的連接key，同時發(fā)現節(jié)點中的所有Source 表都已被連接，節(jié)點中的表還未連接，則此次連接投影的key 列包括列，不再投影保留列，并從索引中刪除節(jié)點，刪除后的索引如圖16 所示。

圖15 連接索引查詢Fig.15 Query by join index

圖16 刪除冗余列及所在索引Fig.16 Delete redundant columns and indexes

3.4 基于投影關系索引的連接順序優(yōu)化

連接順序的優(yōu)化需要基于每次連接的代價計算，包括計算代價和I/O 代價。

對于輸入，估算其表體積()為公式：

其中，()為的元組數，為經過投影關系索引判斷后所保留的列。根據2.1 節(jié)中的傳輸策略介紹，可得RR和BF兩種傳輸策略對應的傳輸代價公式：

在假設和均勻分布的情況下，根據2.2 節(jié)中的連接策略介紹，可得關于Hybrid-Hash-Join 和Sort-Merge-Join 的CPU代價公式及I/O代價公式，其中為分配給Hash Table的內存大小。

據此可以得到CPU 代價公式和I/O 代價公式，其中為選擇因子，1 表示選擇Hash 策略，0 表示選擇Sort-Merge策略。

據此得到每次連接的代價(?)，其中、分別為節(jié)點間傳輸和節(jié)點與本地文件系統間傳輸的I/O 代價權重，為CPU 計算代價權重，三者均與系統硬件相關。

在假設輸入表中連接屬性的值在表中均勻分布的情況下，兩表連接結果的元組數(?) 估算為公式：

其中，為連接鍵值，(,)表示在中不同值的數量。

通過代價公式和連接元組數計算，可以得到基于動態(tài)規(guī)劃的連接順序重排算法4。已知Source 表數量為，初始化狀態(tài)矩陣，將每個Source 表依次加入到中（行4），對于矩陣中的每個節(jié)點，其行編號為（已連接的Source 表數-1），列編號為該連接樹生成時起始的Source 表編號。對于D的連接樹，需要連接的表數共為+1，且包含Source，那么可連接的備選項為之前已計算得出的局部最優(yōu)連接樹D,D(<,<)（行9～10），具體是否可連接，是否為重復連接，通過3.3 節(jié)等多個條件判斷（行11）。通過此方法可以得到D節(jié)點的最優(yōu)連接策略，這樣從到D逐行完成狀態(tài)轉移矩陣，D中的最小代價的連接策略即為結果（行20）。此算法的最差時間復雜度為(),其中為連接表數。

Dynamic programming

輸入：,number of source tables for all connections;,index of the source table.

輸出：,optimal connection diagram.

4 實驗分析

對于本文提出的投影和連接順序優(yōu)化方法，在Flink 系統上進行了實現，通過修改Plan 層的源碼，對比優(yōu)化后與優(yōu)化前的執(zhí)行時間及中間結果傳輸量。通過對比不同體積、不同數量、不同冗余度的連接樹在優(yōu)化前后的執(zhí)行時間和中間數據傳輸量對優(yōu)化方法進行了分析。

4.1 數據集和實驗環(huán)境

本文采用的數據集為TPC-H 標準數據集，TPCH 是TPC 委員會發(fā)布的用于決策支持的測試標準。它由一套面向業(yè)務的即席查詢和并發(fā)數據修改組成，其數據具有廣泛的行業(yè)相關性，其中TPC-H 表的關系模式及表的內容如圖17 所示。本文實驗中所用到的不同規(guī)模的數據集均由TPC-H Tools 指定數據體積生成，其中體積分布如表2 所示。

圖17 TPC-H 關系模式圖Fig.17 Relational schema of TPC-H

表2 TPC-H 表數據體積及分布Table 2 Data size and distribution of TPC-H tables

在5 表到8 表的連接實驗中，默認連接策略皆為優(yōu)先連接體積較大的表，實驗記錄如無特殊聲明并行度皆為36。

4.2 實驗環(huán)境設置

本文所描述的相關技術細節(jié)，均在Flink 1.8.0 中進行實現，采用Java 語言進行編寫。實驗的運行環(huán)境及軟硬件環(huán)境如下。

（1）硬件環(huán)境：實驗采用的分布式環(huán)境由4 臺服務器組成，1 臺主節(jié)點，3 臺從節(jié)點。從節(jié)點采用的服務器配置為：①CPU，Intel Xeon E5-2603 V4 *2，核心數目6 核心，主頻1.7 GHz。②內存64 GB，主頻2 400 MHz。③硬盤400 GB SSD。主節(jié)點配置為：①CPU，Intel Xeon E5-2603 V4 *2，核心數目6 核心，主頻1.7 GHz。②內存128 GB，主頻2 400 MHz。③硬盤400 GB SSD。

（2）軟件環(huán)境：①操作系統Centos 7。②存儲環(huán)境MySQL 5.6.45。③Flink 版本1.8.0，JDK 版本1.8.0。

4.3 實驗結果分析

這里定義一個連接任務中的冗余列c和冗余率，c指此列在此后的連接中不再作為key，且不存在于結果列中，表示所有冗余列數量和占所有Join 表中的列數量的比例。

冗余率為0，即連接性能幾乎只受連接順序影響的情況，不指定數據傳輸策略和連接策略，由系統自動選擇。

如圖18 所示，連接順序優(yōu)化在表數較少的情況下優(yōu)化效果并不明顯，因為低表數下連接順序的解空間同樣非常小，最優(yōu)解與默認解的執(zhí)行效率也很少會出現較大差距。同時可以看出，5個table在700 MB規(guī)模以下的數據集執(zhí)行時間幾乎一致，可以判斷系統啟動執(zhí)行的最小時間為14 s 左右，且5 個table 在700 MB 規(guī)模以下的數據集對系統資源消耗極小，沒有顯著優(yōu)化效果。且通過優(yōu)化前后幾乎無差別的執(zhí)行時間可以看出，5 個table 下的優(yōu)化程序執(zhí)行時間同樣可以忽略不計，即使在小規(guī)模數據集計算上，優(yōu)化程序仍不會對整體運算造成可見的負擔。

圖18 5 表連接的優(yōu)化前后執(zhí)行時間Fig.18 Execution time before and after optimization in 5 tables join

如圖19 所示，在6 個table 下，大規(guī)模數據集對計算時間的影響已非常明顯，并可以看出一定的優(yōu)化效果。由于解空間仍然很小，且已加入的連接表之間的體積差距并不足夠顯著，雖然可以明顯看到優(yōu)化效果，但是優(yōu)化前后的執(zhí)行時間差距并不是很大。

圖19 6 表連接的優(yōu)化前后執(zhí)行時間Fig.19 Execution time before and after optimization in 6 tables join

通過圖20、圖21 可以看出，隨著表數量的增加，優(yōu)化效果變得非常顯著。通過圖22 可以看出，隨著表數量的增加，優(yōu)化效果變得更加明顯，最大達到了58.69%。這是由于新加入的連接尾表體積非常小，大表與小表之間的體積差距非常小，連接順序對連接效率的影響非常顯著，且由于解空間的大小隨著連接表的數量呈階乘級增加，更容易找到效率遠高于默認解的最優(yōu)解。

圖20 7 表連接的優(yōu)化前后執(zhí)行時間Fig.20 Execution time before and after optimization in 7 tables join

圖21 8 表連接的優(yōu)化前后執(zhí)行時間Fig.21 Execution time before and after optimization in 8 tables join

圖22 不同表數下平均執(zhí)行時間優(yōu)化比例Fig.22 Average execution time optimization ratio of different tables

通過圖23～圖26 可以看到，優(yōu)化程序對多表連接的中間數據傳輸有著固定的優(yōu)化效果，由于各節(jié)點的數據傳輸量僅和連接樹有關，不受系統基礎運行時間影響，即使在低連接表數下也可以看到顯著的優(yōu)化效果。通過圖27 可以看到，不同數據集體積下各表有著幾乎一致的優(yōu)化比例，6～8 個table 的連接優(yōu)化比例在不同數據集下的差別均在1%左右浮動。由于本文采用的優(yōu)化算法為確定性算法，對于同一連接樹的優(yōu)化策略是固定的，可以確定對于同一分布的數據集，在相同的優(yōu)化策略下，中間數據傳輸優(yōu)化比例與數據集體積無關。

圖23 5 表連接優(yōu)化前后的中間數據傳輸體積Fig.23 Intermediate data ship size before and after optimization in 5 tables join

圖24 6 表連接優(yōu)化前后的中間數據傳輸體積Fig.24 Intermediate data ship size before and after optimization in 6 tables join

圖25 7 表連接優(yōu)化前后的中間數據傳輸體積Fig.25 Intermediate data ship size before and after optimization in 7 tables join

圖26 8 表連接優(yōu)化前后的中間數據傳輸體積Fig.26 Intermediate data ship size before and after optimization in 8 tables join

通過圖27 與圖28 對比可以得知，中間數據傳輸的優(yōu)化比例遠高于執(zhí)行時間的優(yōu)化比例，這是因為Flink 在執(zhí)行時節(jié)點之間默認采用的為Pipeline 傳輸策略。這個策略規(guī)定了一個節(jié)點產生數據后立刻發(fā)送到下一個節(jié)點進行處理，Pipeline 策略一定程度上可以緩解由于大表在前產生的上游數據阻塞的情況。

圖27 冗余率為0 時的中間數據傳輸體積優(yōu)化比例Fig.27 Intermediate data ship size optimization ratio in redundant ratio 0

圖28 冗余率為0 時的執(zhí)行時間優(yōu)化比例Fig.28 Execution time optimization ratio in redundant 0

這里選取8個table在不同冗余率（=0%，9.70%，19.40%，38.70%，51.60%）下的默認連接策略與優(yōu)化后的連接策略進行對比，由于優(yōu)化程序會清除連接中的冗余列，優(yōu)化后的連接策略可視為冗余率為0。

通過圖29 可以很明顯地看到冗余率的不同對執(zhí)行時間的影響。同樣由于100 MB 體積數據處理時間的優(yōu)化瓶頸不在于連接策略，優(yōu)化比例和500 MB以上的數據存在較大差異。

圖29 不同冗余率下執(zhí)行時間優(yōu)化比例Fig.29 Execution time optimization ratio in different redundant ratios

由于冗余列的存在并不影響計算效率，不同冗余率對運算時間的影響僅來自于中間數據傳輸量對網絡傳輸時間的影響，不同冗余率下執(zhí)行時間優(yōu)化比例明顯和冗余率成正相關，但是彼此間差距并不巨大，冗余率從0%提升到51.60%，而時間優(yōu)化比例僅提升了10%。

通過圖30 可以看出，不同數據集體積下各表在同一冗余率下有著幾乎一致的優(yōu)化比例，這點與前文根據圖27 所得到結論是相符的。在相同的優(yōu)化策略和冗余率下，中間數據傳輸優(yōu)化比例與數據集體積無關。

圖30 不同冗余率下數據傳輸優(yōu)化比例Fig.30 Data ship optimization ratio in different redundant ratios

同時可以看到冗余率對中間數據傳輸的影響要比執(zhí)行時間的影響更大，冗余率從0%到51.60%，優(yōu)化比例最大提升了19%。

表3 對比了0%和9.70%冗余率下500 MB 體積的8 個table 在默認連接策略下的中間節(jié)點的數據發(fā)送量?？梢悦黠@看到冗余列導致了不必要的數據傳輸，且冗余列若不及時通過投影清除會增加后續(xù)節(jié)點無用的數據發(fā)送量。

表3 0%和9.70%冗余率下中間節(jié)點的數據發(fā)送量Table 3 Data ship of intermediate nodes in 0%and 9.70%redundant ratios

圖31 對比了不同并行度下的連接執(zhí)行時間優(yōu)化比例。同樣在100 MB 規(guī)模的小體積數據上優(yōu)化效果并不明顯，隨著數據體積的增大，在數據體積相同時，優(yōu)化比例明顯隨著并行度增加而提高。

圖31 不同并行度下執(zhí)行時間優(yōu)化比例Fig.31 Execution time optimization ratio in different parallelism

5 總結

本文針對大數據連接處理中投影關系的優(yōu)化設計了一種基于投影關系索引的優(yōu)化方法，并結合動態(tài)規(guī)劃算法在Flink 中進行了實現。實驗結果表明，本文方法所針對的冗余列問題確實對連接效率有著顯著影響，隨著冗余率的增加，連接效率會明顯降低。本文所提供的優(yōu)化方法在優(yōu)化連接順序的同時考慮到了冗余列的影響，可以大大減少連接任務的執(zhí)行時間和中間數據傳輸量。