張 磊， 周敏奇， 王 立

（華東師范大學(xué)軟件學(xué)院，上海 200062）

1 引 言

隨著Web 2.0時代的到來，大數(shù)據(jù)處理已然成為學(xué)術(shù)界和工業(yè)界研究的熱點(diǎn)問題.OLAP系統(tǒng)在決策支持和數(shù)據(jù)挖掘領(lǐng)域扮演著越來越關(guān)鍵的角色，BI形式的復(fù)雜查詢變得更加頻繁.比如，上海證券交易所對近三個月的股票交易記錄作分析，中國鐵路總公司12306網(wǎng)站對實(shí)時成千上萬并發(fā)查詢的支持，交通銀行對用戶查詢最近業(yè)務(wù)信息的支持，無一不需要底層高性能數(shù)據(jù)庫操作做保證.

近幾年來，計算機(jī)硬件發(fā)展迅猛，尤其多核CPU和大容量內(nèi)存技術(shù)的進(jìn)步，1 MB內(nèi)存的價格由1980年的1萬美元降低到2010年的0.01美元，SUN公司也提出將來服務(wù)器中的CPU核數(shù)將上升至32到128個之多，硬件的變革使得內(nèi)存計算成為提高數(shù)據(jù)庫查詢性能的新理念.雖然如此，但整個數(shù)據(jù)庫的表數(shù)據(jù)都存儲在一臺服務(wù)器的內(nèi)存中仍然是無法實(shí)現(xiàn)的，隨著國內(nèi)工業(yè)界正興起的去IOE運(yùn)動，集中式計算為代表的小型機(jī)逐步被淘汰，分布式環(huán)境將會在各大公司數(shù)據(jù)中心成為標(biāo)準(zhǔn)配置.由于業(yè)務(wù)需要，分布式內(nèi)存計算系統(tǒng)日漸流行，由此也引發(fā)了對分布式內(nèi)存數(shù)據(jù)庫系統(tǒng)中操作算子的重新設(shè)計.

分布式內(nèi)存計算環(huán)境下的哈希連接算法涉及混洗［4］和連接等多個操作算子.首先，從分布式文件系統(tǒng)中分別讀取待連接的兩張表；然后經(jīng)過數(shù)據(jù)重分區(qū)將兩張表中連接屬性哈希值相同的記錄傳輸?shù)较嗤?jié)點(diǎn)；最后兩張表屬性哈希值相同的數(shù)據(jù)在單節(jié)內(nèi)部做連接.在此過程中，混洗涉及網(wǎng)絡(luò)數(shù)據(jù)傳輸，由于網(wǎng)絡(luò)傳輸速率的限制，混洗操作成為系統(tǒng)中算法執(zhí)行的性能瓶頸，各種分布式系統(tǒng)都不斷優(yōu)化混洗操作的實(shí)現(xiàn)細(xì)節(jié).已有的Hadoop［1］系統(tǒng)和Spark［2－12］系統(tǒng)雖然對混洗過程都有所優(yōu)化，但是它們并沒有對兩張表連接屬性上的數(shù)據(jù)分布做精確統(tǒng)計.本文所提出的LCDJ算法在優(yōu)化網(wǎng)絡(luò)傳輸上起到了較好效果.在既定的數(shù)據(jù)分布環(huán)境下，LCDJ算法精確統(tǒng)計待連接表在其連接屬性上的屬性值分布，進(jìn)而構(gòu)建以處理時間為開銷的代價模型，通過選擇節(jié)點(diǎn)生成優(yōu)化連接方案，大幅減少了時間開銷.本文貢獻(xiàn)為：（1）建立了以處理時間為開銷，考慮網(wǎng)絡(luò)傳輸代價、并行度和計算負(fù)載均衡的代價模型；（2）相對于枚舉方法選擇連接節(jié)點(diǎn)排列，提出一種速度更快的次優(yōu)方法選擇次優(yōu)節(jié)點(diǎn)排列；（3）改進(jìn)了數(shù)據(jù)混洗操作和單機(jī)節(jié)點(diǎn)內(nèi)部的連接算法.

本文組織如下，第二節(jié)介紹相關(guān)工作；第三節(jié)將給出問題定義；第四節(jié)詳細(xì)論述LCDJ算法，且建立代價模型分析算法各個階段的性能消耗；第五節(jié)將介紹基于原型系統(tǒng)的連接算法實(shí)驗(yàn)和相關(guān)性能分析；第六節(jié)總結(jié)全文.

2 相關(guān)工作

內(nèi)存技術(shù)快速發(fā)展，近幾年基于內(nèi)存計算的數(shù)據(jù)庫和分布式系統(tǒng)被廣泛使用.關(guān)系型數(shù)據(jù)庫中，SAP HANA［3］以半連接大幅提升連接性能，Hyper以NEO連接［5］優(yōu)化網(wǎng)絡(luò)調(diào)度來提升連接性能.在Nosql領(lǐng)域，基于內(nèi)存計算的Spark生態(tài)圈異軍突起，基于Spark內(nèi)核的數(shù)據(jù)倉庫基礎(chǔ)架構(gòu)Shark［13］，它提供兩種分布式連接算法，分別為map連接和shuffle連接，相當(dāng)于Hive系統(tǒng)中的map端連接和reduce端連接.map連接是將小表廣播傳輸?shù)剿泻写蟊矸謪^(qū)的節(jié)點(diǎn)上，shuffle連接則將兩個表都按照相同哈希函數(shù)在連接屬性上做切分，然后在第三批節(jié)點(diǎn)上做連接，但是這兩種連接均未使用精確統(tǒng)計數(shù)據(jù)分布方式重點(diǎn)優(yōu)化網(wǎng)絡(luò)傳輸代價.

在單節(jié)點(diǎn)內(nèi)部，多核大內(nèi)存計算環(huán)境使得傳統(tǒng)獲取數(shù)據(jù)引發(fā)的磁盤IO代價已幾乎不存在，但放大了內(nèi)存和CPU之間的內(nèi)存墻問題.在Monetdb中，radix連接［6］算法考慮了內(nèi)存連接算法中的Cache缺失和TLB缺失，充分利用了硬件特征.但Blanas S在他的論文［7］中提到，基于內(nèi)存的哈希連接算法在省略了radix連接中分區(qū)階段的時候，不會比包括radix連接在內(nèi)的引入分區(qū)階段的哈希算法差，同時由于沒有分區(qū)階段，無需考慮硬件參數(shù)反而能夠充分利用硬件的預(yù)取和簡化向查詢優(yōu)化器傳參.但是Balkesen C在論文［8］實(shí)驗(yàn)中證明radix連接為代表的考慮硬件參數(shù)優(yōu)化的連接算法還是比無分區(qū)階段的連接算法性能更優(yōu)，他指出Blanas S的論文中利用小表建立哈希表，減小了分區(qū)階段在整個算法中的占比，低估了radix連接的性能.除了哈希連接算法，Chhugani J則在他的論文［9］中說明了利用SIMD可以加速循環(huán)嵌套連接算法.此外，內(nèi)存中的排序歸并連接算法也是充分利用了硬件特征，在Kim C的論文［10］中，由于SIMD指令集可以加速排序，他提出在SIMD寬度超過256位的時候，排序歸并連接可以超越哈希連接的性能，但是Albutiu M C提出的MPSM［11］并行排序歸并算法則考慮NUMA架構(gòu)對算法的影響，并且強(qiáng)調(diào)MPSM可以在沒有SIMD的支持下性能比哈希連接算法好.

本文提出的LCDJ算法本質(zhì)上還是shuffle連接，但是在連接操作之前還增加了統(tǒng)計數(shù)據(jù)分布階段和篩選連接節(jié)點(diǎn)排列階段，單機(jī)節(jié)點(diǎn)內(nèi)部采用無分區(qū)哈希連接算法.

3 問題定義

在商用大內(nèi)存PC服務(wù)器組成的分布式環(huán)境下，兩張關(guān)系表數(shù)據(jù)分片存儲于多個節(jié)點(diǎn)的內(nèi)存中，和傳統(tǒng)單機(jī)基于磁盤的數(shù)據(jù)庫連接算法有如下不同：首先在內(nèi)存環(huán)境中，磁盤IO消耗可以忽略；然后，在分布式環(huán)境下，網(wǎng)絡(luò)傳輸代價將成為瓶頸，必須作為一個新的因素加入到代價建模中.本文的研究重點(diǎn)為提升單個連接算法性能，后續(xù)工作將考慮容錯和其他查詢的影響.

考慮分布式內(nèi)存計算環(huán)境下的哈希連接算法，其中一個典型連接場景為，對于R（a，b）和S（c，d）兩張關(guān)系表，R表存儲在m 個節(jié)點(diǎn)上（每個節(jié)點(diǎn)上含有分片r1，r2，r3，...，rm－1，rm，m≥1），S表存儲在n 個節(jié)點(diǎn)上（每個節(jié)點(diǎn)上含有分片s1，s2，s3，...，sn－1，sn，n≥1），令R表和S表在a和c屬性上做連接：

依照傳統(tǒng)數(shù)據(jù)庫做法，以上給定的SQL查詢先由SQL解析器解析為抽象語法樹，然后經(jīng)過分析優(yōu)化生成邏輯計劃，再由邏輯計劃生成物理執(zhí)行計劃得以執(zhí)行.

圖1 邏輯計劃及物理執(zhí)行計劃Fig.1 Logical plan and physical execution plan

圖1所示的邏輯計劃在R表上依次為scan操作和filter操作，在S表上為scan操作，最終做join操作.與單機(jī)數(shù)據(jù)庫不同的是，當(dāng)此邏輯計劃轉(zhuǎn)化為物理執(zhí)行計劃時，由于原始數(shù)據(jù)沒有被切分，join操作的兩個孩子都必須加上exchange操作.exchange操作類似于mapreduce中的混洗，但在我們原型系統(tǒng)中的exchange操作無需將中間結(jié)果持久化到磁盤上，且exchange的網(wǎng)絡(luò)傳輸速度達(dá)到網(wǎng)絡(luò)極限速度.為了充分減少網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量，同時考慮并行度和計算負(fù)載因素，現(xiàn)將問題定義為：選出節(jié)點(diǎn)排列C（c1，c2，…，cT，T≥1）作為連接節(jié)點(diǎn)，使下文中定義的算法連接算法執(zhí)行時間Tsum最小.

4 LCDJ算法

現(xiàn)在數(shù)據(jù)中心通常采用千兆網(wǎng)交換機(jī)，服務(wù)器也通常配置千兆網(wǎng)卡，網(wǎng)絡(luò)傳輸速度一般為125 MB／s，嚴(yán)重制約分布式系統(tǒng)的性能.所以減小均網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量（下文中稱作網(wǎng)絡(luò)傳輸數(shù)據(jù)量）能夠在性能提升上起決定性作用，與此同時，并行度和由數(shù)據(jù)傾斜產(chǎn)生的計算負(fù)載均衡問題，也是LCDJ算法優(yōu)化的重點(diǎn).

LCDJ算法分為三個階段：（1）統(tǒng)計各個含有數(shù)據(jù)的節(jié)點(diǎn)上連接屬性的分布，將數(shù)據(jù)分布發(fā)送至主節(jié)點(diǎn)，直到主節(jié)點(diǎn)接收到所有含有數(shù)據(jù)的節(jié)點(diǎn)傳輸來的數(shù)據(jù)分布為止，此階段時間為Thist；（2）主節(jié)點(diǎn)根據(jù)每個含有數(shù)據(jù)的節(jié)點(diǎn)傳輸來的數(shù)據(jù)分布，利用本文提出的篩選算法計算出使性能接近最優(yōu)的連接節(jié)點(diǎn)排列，此階段時間為Toptim；（3）確定連接節(jié)點(diǎn)排列后，從掃描兩張表的數(shù)據(jù)到最后連接完成，此階段時間為Tjoin.在沒有前兩步優(yōu)化前，分布式哈希連接操作執(zhí)行的時間為Torig.現(xiàn)有：

LCDJ算法將建立以時間為開銷的代價模型，考慮數(shù)據(jù)傳輸量，并行度和計算負(fù)載均衡.目標(biāo)為最小化分布式連接操作的時間Tsum，并且使Tsum相對于Torig節(jié)省的時間開銷比Thist＋Toptim大的多.LCDJ算法將分別在各階段中采用優(yōu)化策略，減小每個階段的時間代價，最終使得Tsum盡可能小.

4.1 統(tǒng)計分布

LCDJ算法第一個階段為統(tǒng)計數(shù)據(jù)分布，當(dāng)一個足夠大的數(shù)據(jù)庫表文件分布式地存儲于集群節(jié)點(diǎn)中時，主節(jié)點(diǎn)可以利用各節(jié)點(diǎn)連接屬性上的數(shù)據(jù)分布精確統(tǒng)計網(wǎng)絡(luò)傳輸開銷.

統(tǒng)計數(shù)據(jù)分布有兩種方法：抽樣統(tǒng)計和直方圖統(tǒng)計.前者比后者掃描的數(shù)據(jù)更少，一般統(tǒng)計10%的數(shù)據(jù)即能達(dá)到較好效果，但后者比前者更精確.由于內(nèi)存數(shù)據(jù)庫選用的機(jī)器一般都有多核且大內(nèi)存的特點(diǎn)，再者，除了修改統(tǒng)計值需要加鎖，統(tǒng)計數(shù)據(jù)分布只需單純地讀內(nèi)存操作，完全可以采用多線程統(tǒng)計直方圖.還有一個因素是由于網(wǎng)絡(luò)傳輸速度是瓶頸，在掃描數(shù)據(jù)的時候精確統(tǒng)計從而避免對網(wǎng)絡(luò)傳輸數(shù)據(jù)量的計算不精確.

直到每個含有相關(guān)數(shù)據(jù)的節(jié)點(diǎn)發(fā)送的直方圖都到達(dá)主節(jié)點(diǎn)，第一階段結(jié)束，此階段花費(fèi)的時間為Thist.第一階段主要是為了讓主節(jié)點(diǎn)能夠察覺兩張表數(shù)據(jù)的精確分布信息，以便為隨后的篩選優(yōu)化提供數(shù)據(jù)參考.

4.2 篩選優(yōu)化

LCDJ算法第二階段為根據(jù)第一階段獲得的精確統(tǒng)計信息，快速計算出合適的節(jié)點(diǎn)集合排列作為連接節(jié)點(diǎn).首先，由于網(wǎng)絡(luò)傳輸速度成為瓶頸，所傳輸?shù)臄?shù)據(jù)量盡量小是本算法優(yōu)化的目標(biāo).其次由于系統(tǒng)調(diào)度任務(wù)也會占用時間開銷，并且此時間開銷在內(nèi)存計算系統(tǒng)中會比其他基于磁盤的分布式系統(tǒng)占更大比重，所以并行度也是影響性能的一個因素.再次，連接節(jié)點(diǎn)的計算負(fù)載均衡情況也是隨著節(jié)點(diǎn)排列的變化而改變.

所以節(jié)點(diǎn)排列的選擇受到以上三個因素的影響，下文分別分析這三個因素并建立代價模型.最終目的是篩選出最優(yōu)節(jié)點(diǎn)排列作為分布式連接執(zhí)行的連接節(jié)點(diǎn).

根據(jù)前面問題定義中的描述，現(xiàn)定義代價模型中將要用到的參數(shù).兩張表R和S中記錄大小分別為rtuplesize和stulpesize，集群帶寬為vnet.集群當(dāng)中有k個節(jié)點(diǎn)，兩表各個分片所在的節(jié)點(diǎn)由nodeof（）得出，例如ri所在的節(jié)點(diǎn)編號為nodeof（ri），此編號從1開始.獲得R表各分區(qū)的精確統(tǒng)計數(shù)據(jù)hist（ri）（i＝1，2，3，…，m－1，m，1≤m≤k），和S表各分區(qū)的精確統(tǒng)計數(shù)據(jù)為hist（sj）（j＝1，2，…，n－1，n，1≤n≤k），m 和n分別為兩張表的分片數(shù)量.直方圖桶的個數(shù)為BKS.我們的目的是選擇出一個節(jié)點(diǎn)排列C（c1，c2，…，cT，1≤T≤k）使得代價cost最小，節(jié)點(diǎn)排列和節(jié)點(diǎn)集合不同，節(jié)點(diǎn)排列考慮節(jié)點(diǎn)集合根據(jù)哈希值有特定的順序.以下分析也均是在針對特定的節(jié)點(diǎn)排列C來做代價建模，cost將在下文中給出.

網(wǎng)絡(luò)傳輸數(shù)據(jù)量：篩選優(yōu)化過程中網(wǎng)絡(luò)傳輸數(shù)據(jù)量由兩個表分別所在的各個節(jié)點(diǎn)需要傳輸?shù)臄?shù)據(jù)量之和與最終做連接節(jié)點(diǎn)的個數(shù)決定.

一旦某個含有相關(guān)數(shù)據(jù)的節(jié)點(diǎn)被選為最終的連接節(jié)點(diǎn)，此節(jié)點(diǎn)上對應(yīng)哈希段的數(shù)據(jù)不需要被傳輸，如圖2虛線部分，node1上哈希值為0的記錄不需要傳輸，同理node3上哈希為3的記錄不需要傳輸.

由此可得出未被選擇作為連接節(jié)點(diǎn)時，需要傳輸?shù)臄?shù)據(jù)量是此節(jié)點(diǎn)上所有的相關(guān)數(shù)據(jù)，例如對ri來說，其所在節(jié)點(diǎn)上所有的數(shù)據(jù)量為

如果t＝nodeof（ri）被選為連接節(jié)點(diǎn)，不用傳輸?shù)臄?shù)據(jù)量的大小為

圖2 重分區(qū)哈希連接Fig.2 Repartition hash join

以上為ri所在節(jié)點(diǎn)的傳輸量，可得出連接節(jié)點(diǎn)排列C中ci接收R表數(shù)據(jù)量的平均值為：

trsR的取值雖然不是單調(diào)，但是在一定區(qū)間范圍內(nèi)選擇的連接節(jié)點(diǎn)個數(shù)越多，平均網(wǎng)絡(luò)傳輸量就越小.但總體時間開銷并非隨節(jié)點(diǎn)個數(shù)增多而減小，因?yàn)橄到y(tǒng)需要調(diào)度的任務(wù)個數(shù)也會增大，所以并行度因素同樣影響性能，下面我們將并行度因素考慮進(jìn)模型中.

并行度 系統(tǒng)將連接任務(wù)序列化直到傳輸?shù)竭B接節(jié)點(diǎn)上，這稱作一次調(diào)度任務(wù).任務(wù)越復(fù)雜，序列化和網(wǎng)絡(luò)傳輸所花費(fèi)的時間代價越大.加上內(nèi)存計算系統(tǒng)處理內(nèi)存中數(shù)據(jù)的速度較磁盤快，所以作業(yè)調(diào)度時間比重比基于磁盤系統(tǒng)的調(diào)用時間更大.調(diào)度一次任務(wù)的時間為s，由于需要調(diào)度的任務(wù)數(shù)由節(jié)點(diǎn)排列C中連接節(jié)點(diǎn)個數(shù)決定，所以本算法在任務(wù)調(diào)度上所需要的時間為s＊T.

計算負(fù)載均衡 由于基于內(nèi)存計算的系統(tǒng)計算速度相對更快，因此負(fù)載均衡更加重要.最終選擇出的連接節(jié)點(diǎn)對于表R的計算負(fù)載平均值為：

故，當(dāng)nodeof（ri）∈C 時，有

當(dāng)nodeof（Ri）?C時，有

joinri為在節(jié)點(diǎn)排列C中的第i個節(jié)點(diǎn)上對于表R的計算負(fù)載：

可得出連接節(jié)點(diǎn)相對于表R的計算負(fù)載均衡因子bR為：

由于需要考慮以上三點(diǎn)因素，根據(jù)系統(tǒng)中參數(shù)設(shè)定，將三個因素分別用時間來衡量且分別乘以一個權(quán)重因子，這里權(quán)重因子由系統(tǒng)調(diào)參給出.得出最終代價的等式為

可見以上代價模型中，影響整個代價的變量為所選連接節(jié)點(diǎn)集合以及其順序，即節(jié)點(diǎn)排列C.以上代價分析中是將特定的集合C帶入最終求得cost的值，但如何求得最優(yōu)代價對應(yīng)的集合C，下面給出兩種方案，目的是在滿足cost接近最小化的同時使Toptim最小.

枚舉方法：連接節(jié)點(diǎn)排列受節(jié)點(diǎn)個數(shù)和順序影響，枚舉方法可以精確地選擇出最優(yōu)節(jié)點(diǎn)排列，在一個含有k個節(jié)點(diǎn)的集群中，總會含有一個使代價模型最優(yōu)的連接節(jié)點(diǎn)排列.但是枚舉過程需要列出的可能情況，有種，枚舉出這其中的每一種情況，然后帶入cost計算出代價，最終比較得出最小代價對應(yīng)的節(jié)點(diǎn)排列，這是一個時間復(fù)雜度非常大的計算過程，在兩個表數(shù)據(jù)量很大或k值很大的情況下不可能完成.

次優(yōu)方法：雖然枚舉方法是必然能夠產(chǎn)生最優(yōu)解的算法，但枚舉方法的時間復(fù)雜度非常大.由于大表連接通常涉及節(jié)點(diǎn)數(shù)量較多，故只能逼近最優(yōu)解.在此我們采用先確定節(jié)點(diǎn)個數(shù)再確定節(jié)點(diǎn)排列的方法來求解.經(jīng)過對cost等式的分析，等式最終可以近似簡化為：

在此，p和q分別可以表示為

步驟一：確定節(jié)點(diǎn)個數(shù).由于本算法的cost可以簡化為px＋q／x的形式，由于其函數(shù)特性為先減后增，故cost有個最小值.并且當(dāng)x在附近取值時，節(jié)點(diǎn)個數(shù)取值將在t1＝

由于q的值確定，所以x的取值范圍完全由p決定，p的值由負(fù)載均衡和傳輸數(shù)據(jù)量決定.當(dāng)完全負(fù)載均衡且傳輸數(shù)據(jù)量最小時，p取值最小，完全負(fù)載均衡時，（bR＋bS）／vnet為0，在此近似定義傳輸數(shù)據(jù)量最小值為節(jié)點(diǎn)中所有數(shù)據(jù)減去最大數(shù)據(jù)量節(jié)點(diǎn)上的數(shù)據(jù).當(dāng)負(fù)載均衡最差，且傳輸數(shù)據(jù)量最大時，p取值最大，負(fù)載均衡最差為當(dāng)所有數(shù)據(jù)均集中在一個節(jié)點(diǎn)上，在此近似定義傳輸數(shù)據(jù)量最大值為節(jié)點(diǎn)中所有數(shù)據(jù)減去最小數(shù)據(jù)量節(jié)點(diǎn)上的數(shù)據(jù).

步驟二：確定節(jié)點(diǎn)排列.步驟一確定了節(jié)點(diǎn)個數(shù)所在的一個小區(qū)間，這個區(qū)間內(nèi)含有若干個節(jié)點(diǎn)個數(shù).對于每個節(jié)點(diǎn)個數(shù)，首先統(tǒng)計針對具體連接節(jié)點(diǎn)個數(shù)為模的數(shù)據(jù)分布，然后優(yōu)先選擇相應(yīng)哈希段數(shù)據(jù)量較大的留做相應(yīng)的連接節(jié)點(diǎn).

圖3左部為各節(jié)點(diǎn)上數(shù)據(jù)分布的統(tǒng)計結(jié)果.首先node1和node3中有最大的哈希段，分別為node1中的0哈希段和node3中的3哈希段，因此首先將node1作為第0個節(jié)點(diǎn)，將node3作為第3個節(jié)點(diǎn).然后較大的是node2中的2哈希段，因此將node2選為第2個節(jié)點(diǎn).最后將node4選為第1個節(jié)點(diǎn).該策略在較小節(jié)點(diǎn)區(qū)間內(nèi)主要考慮傳輸數(shù)據(jù)量因素，來確定最終連接節(jié)點(diǎn)的排列.選出的節(jié)點(diǎn)排列可能無法使執(zhí)行性能達(dá)到最優(yōu)，但篩選過程的時間消耗可以保持在可控范圍內(nèi).

4.3 分布式哈希連接

LCDJ算法第三階段為分布式哈希連接算法執(zhí)行，本階段采用前兩個階段篩選的節(jié)點(diǎn)排列作為連接節(jié)點(diǎn)排列.另外在此階段還采用了針對分布式連接優(yōu)化的多項(xiàng)技術(shù)，比如按塊執(zhí)行，流水線式exchange和單節(jié)點(diǎn)無分區(qū)哈希連接.

圖3 確定節(jié)點(diǎn)排列步驟Fig.3 Determine nodes’permutation

按塊執(zhí)行對數(shù)據(jù)庫查詢這種針對成千上萬數(shù)據(jù)頻繁調(diào)用相同函數(shù)的進(jìn)程來說，盡量減少函數(shù)的調(diào)用次數(shù)將對性能有本質(zhì)的提升，按記錄執(zhí)行和按塊執(zhí)行在函數(shù)調(diào)用次數(shù)上，后者顯然較少.此外在基于內(nèi)存的計算系統(tǒng)中數(shù)據(jù)高局部性能減少cache缺失的次數(shù).

流水線式exchange exchange會將兩張表相同哈希段內(nèi)的記錄都傳輸?shù)酵还?jié)點(diǎn)上，完全避免不同節(jié)點(diǎn)之間的數(shù)據(jù)依賴性.LCDJ算法實(shí)現(xiàn)的exchange操作類似于mapreduce中的shuffle，但又有所不同.mapreduce中的shuffle需要在map端持久化，而exchange是非阻塞的，也就是直接將接收的數(shù)據(jù)做哈希然后傳輸?shù)侥繕?biāo)節(jié)點(diǎn)，可以看做是節(jié)點(diǎn)內(nèi)存之間的傳輸，同時exchange還被實(shí)現(xiàn)為防止數(shù)據(jù)傾斜，使網(wǎng)絡(luò)帶寬穩(wěn)定維持在最大利用狀態(tài).實(shí)驗(yàn)證明，exchange如果將map端的數(shù)據(jù)持久化到磁盤后傳輸，遠(yuǎn)不及直接傳輸?shù)乃俣?，后者是前者速度?－3倍，對于作為瓶頸的網(wǎng)絡(luò)帶寬來說，exchange操作使網(wǎng)絡(luò)帶寬達(dá)到極限，也與我們前面的代價模型相符.

單節(jié)點(diǎn)無分區(qū)哈希連接在單節(jié)點(diǎn)內(nèi)部采用無分區(qū)哈希連接算法［7］，雖然radix連接［6］是能夠針對硬件配置優(yōu)化的高效連接算法，但radix連接在本算法中并不適用.首先因?yàn)閞adix連接的前提是數(shù)據(jù)既已存在于內(nèi)存中，而本算法中數(shù)據(jù)是不斷地傳輸?shù)竭B接節(jié)點(diǎn)上；其次因?yàn)楸舅惴ǖ钠款i和radix連接要克服的瓶頸不一致，本算法的瓶頸在于網(wǎng)絡(luò)傳輸帶寬，但radix連接的瓶頸在于內(nèi)存帶寬遠(yuǎn)遠(yuǎn)小于CPU處理頻率.

由于radix連接不是很適合作為我們算法中的單機(jī)連接算法，故在此采用無分區(qū)并行哈希連接算法，build階段多個線程（圖4中為四線程）遍歷R表建立共享哈希表，probe階段遍歷S表匹配哈希表中的記錄.

圖4 無分區(qū)哈希連接算法Fig.4 No partition hash join algorithm

圖4中，哈希表在哈希連接算法中的實(shí)現(xiàn)非常關(guān)鍵，直接決定連接操作的性能.在哈希表的實(shí)現(xiàn)中，有三個方面的優(yōu)化：（1）build階段哈希表加鎖的粒度為桶.圖4雖然是共享哈希表，四個線程同時向哈希表中寫記錄時，可能會發(fā)生同一時刻寫入同一個桶的情形.其中一個線程每次寫哈希表時，對整張哈希表加鎖是一種方法，但一定會產(chǎn)生性能的極大消耗，如果對每個桶都維護(hù)一個鎖，加鎖粒度變?yōu)橥埃軌驑O大減小插入沖突時加鎖的時間；（2）盡量增大哈希表桶的個數(shù)；增大哈希表桶的個數(shù)，使每個桶中的記錄盡量少，可以減小兩個線程向同一個桶中寫記錄的概率，在probe階段，掃描單個桶的代價也會降低；（3）鎖的類型我們使用自旋鎖；在多核環(huán)境下，高并發(fā)度引發(fā)加鎖頻繁，同時由于桶個數(shù)增加而使競爭很少發(fā)生，所以使用自旋鎖比互斥鎖更合適，總開銷小.

5 實(shí)驗(yàn)評估

本文所有實(shí)驗(yàn)均在10個節(jié)點(diǎn)組成的無共享集群上完成的.集群通過千兆以太網(wǎng)交換機(jī)連接.每個節(jié)點(diǎn)有24 GB內(nèi)存，兩個6核CPU，超線程個數(shù)達(dá)到24個.CPU型號為Intel（R）Xeon（R）CPU E5－2620，CPU每個核的頻率為2.00GHz.機(jī)器上運(yùn)行的是RedHat6.4系統(tǒng)，內(nèi)核版本號為2.6.32－358.el6.x86＿64，系統(tǒng)中采用的gcc版本為4.4.7，g＋＋版本為

4.4.7，編譯階段使用O3優(yōu)化.

本文實(shí)驗(yàn)分為三部分，第一部分對LCDJ算法中的三個階段進(jìn)行單獨(dú)實(shí)驗(yàn)并且評估：（1）統(tǒng)計數(shù)據(jù)分布采用抽樣、直方圖或者多線程直方圖方式；（2）計算篩選節(jié)點(diǎn)用枚舉方法和次優(yōu)方法在不同規(guī)模集群下的計算代價；（3）分析對于不同節(jié)點(diǎn)個數(shù)，各個因素對代價的影響.第二部分為篩選出最優(yōu)的10個節(jié)點(diǎn)排列以及連接操作所花的時間.第三部分將對比LCDJ算法和Hive、Shark系統(tǒng)對問題定義中提出的相同的查詢的性能.實(shí)驗(yàn)所采用的數(shù)據(jù)為模式分別為R（int，int）和S（int，int）的兩張表，表數(shù)據(jù)集分為三類：350 MB＊350 MB為兩張45 869 600個元組的表做連接，隨機(jī)生成表數(shù)據(jù)時，是將隨機(jī)數(shù)模上45 869 600；700 MB＊700 MB為兩張91 750 400個元組的表做連接，隨機(jī)生成數(shù)據(jù)時，是將隨機(jī)數(shù)模上91 750 400；1.4 GB＊1.4 GB為兩張183 478 400個元組的表做連接，隨機(jī)生成數(shù)據(jù)時，是將隨機(jī)數(shù)模上183 478 400.

5.1 各階段分析

5.11 數(shù)據(jù)統(tǒng)計

首先對第一個階段進(jìn)行實(shí)驗(yàn)分析，即統(tǒng)計各個節(jié)點(diǎn)上數(shù)據(jù)分布所花費(fèi)的平均時間，分別對節(jié)點(diǎn)上有350 MB、700 MB和1.4 GB的數(shù)據(jù)統(tǒng)計分布.

實(shí)驗(yàn)中，抽樣的比例為10%，抽樣方法由于掃描內(nèi)存中的數(shù)據(jù)量小是最快方法，但從上圖可以看出多線程直方圖方法可以達(dá)到抽樣方法的性能，雖然有一點(diǎn)差距，但這點(diǎn)差距相對于隨后的兩個階段的時間開銷非常小.

5.1.2 篩選時間

枚舉方法和次優(yōu)方法的主要時間差距受節(jié)點(diǎn)個數(shù)的影響.下面是當(dāng)節(jié)點(diǎn)個數(shù)不同時，兩種方法的時間消耗情況.

上圖說明，在節(jié)點(diǎn)達(dá)到7個的時候，枚舉方法的時間開銷已經(jīng)達(dá)到秒級.一旦節(jié)點(diǎn)個數(shù)再增大，時間開銷會大幅增長以致無法執(zhí)行算法.但次優(yōu)方法篩選節(jié)點(diǎn)所需時間隨著節(jié)點(diǎn)增多并沒有增長.

圖5 三種方法統(tǒng)計時間Fig.5 Statistic time of the three methods

圖6 篩選時間比較Fig.6 Comparison of select－times

5.1.3 節(jié)點(diǎn)個數(shù)影響

下圖是針對于4.2中cost所得的每個節(jié)點(diǎn)集合的最優(yōu)情況，采用700 MB＊700 MB數(shù)據(jù)集.其中每種情況對應(yīng)一種節(jié)點(diǎn)排列.

圖7 節(jié)點(diǎn)個數(shù)影響Fig.7 Effect of number of nodes

可見，三種因素中起決定性作用的為均傳輸量.隨著節(jié)點(diǎn)的增多，均傳輸量變小，但是同時調(diào)度總時間在上升，計算負(fù)載均衡所換算出的時間并沒有規(guī)律.圖中綜合三個方面的因素，當(dāng)節(jié)點(diǎn)個數(shù)為9個的時候，其中一個節(jié)點(diǎn)排列會有最優(yōu)的性能.

5.2 篩選節(jié)點(diǎn)

下圖為時間性能最優(yōu)的十種節(jié)點(diǎn)排列，采用的是700 MB＊700 MB數(shù)據(jù)集.排列編號對應(yīng)關(guān)系為：

表1 節(jié)點(diǎn)排列編號對照Tab.1 Nodes’permutation reference

圖8 篩選節(jié)點(diǎn)比較Fig.8 Comparison of select－nodes

可見，最終時間代價最小值所對應(yīng)的排列并不是在最小或最大的節(jié)點(diǎn)個數(shù)上，排列7和排列8節(jié)點(diǎn)個數(shù)相同，但是其兩者時間代價依然會有所不同，且排列7對應(yīng)的時間代價會更小.

5.3 對比實(shí)驗(yàn)

與現(xiàn)在應(yīng)用較廣泛的兩個數(shù)據(jù)查詢系統(tǒng)做對比，分別為Hive 0.12.0和Shark 0.9.

圖9 與Hive、Shark對比Fig.9 Comparison with Hive and Shark

Hive在該查詢之上性能最差，由于Hive基于mapreduce，其物理執(zhí)行計劃被轉(zhuǎn)化為MR作業(yè)，在每次MR作業(yè)之后都有結(jié)果數(shù)據(jù)被寫到HDFS上，所以其速度比較慢.Shark較快，原因是Shark用PDE優(yōu)化了shuffle，但Shark的優(yōu)化較為粗糙，沒有充分考慮傳輸數(shù)據(jù)量的優(yōu)化，而這恰恰是瓶頸.本文提出LCDJ算法較前兩者性能更優(yōu)，其中得益于對數(shù)據(jù)的精確統(tǒng)計和篩選連接節(jié)點(diǎn)排列.

6 總 結(jié)

本文主要研究分布式內(nèi)存計算環(huán)境下的哈希連接算法.提出了一種基于局部感知的三階段算法LCDJ，結(jié)合連接節(jié)點(diǎn)均傳輸數(shù)據(jù)量、并行度以及計算負(fù)載均衡三個因素建立代價模型.利用次優(yōu)方法篩選連接節(jié)點(diǎn)排列使代價接近最小，然后采用非阻塞exchange操作傳輸數(shù)據(jù)，最終在單個節(jié)點(diǎn)下無分區(qū)哈希連接兩張表的數(shù)據(jù).實(shí)驗(yàn)結(jié)果表明，篩選出最優(yōu)節(jié)點(diǎn)排列確實(shí)能夠使得在增加前兩個步驟的前提下，性能依然優(yōu)于未篩選前，且優(yōu)于Hive和Shark兩個系統(tǒng)在相同查詢下的性能.

［1］ http：／／hadoop.apache.org.

［2］ http：／／spark.apache.org.

［3］ 普拉特納H，蔡爾A.內(nèi)存數(shù)據(jù)管理［M］.SAP譯.1版.北京：清華大學(xué)出版社，2013.

［4］ DEAN J，GHENMAWAT S.MapReduce：simplified data processing on large clusters［J］.Communications of the ACM，2008，51（1）：107－113.

［5］ R?DIGER W，MüHLBAUER T，UNTERBRUNER P，et al.Locality－sensitive operators for parallel main－memory database clusters［C］／／IEEE International Conference on Data on Engineering，ICDE，2014.

［6］ BONCZ P A，KERSTEN M L，MANEGOLD S.Breaking the memory wall in MonetDB［J］.Communications of the ACM，2008，51（12）：77－85.

［7］ BLANAS S，LI Y，PATEL J M.Design and evaluation of main memory hash join algorithms for multi－core CPUs［C］／／Proceedings of the 2011 ACM SIGMOD International Conference on Management of data.ACM，2011，37－48.

［8］ BALKESEN C，TEUBNER J，ALONSO G，et al.Main－memory hash joins on multi－core CPUs：Tuning to the underlying hardware［C］／／Data Engineering（ICDE），2013 IEEE29th International Conference on.IEEE，2013：362－373.

［9］ CHHUGANI J，NGUYEN A D，LEE V W，et al.Efficient implementation of sorting on multi－core SIMD CPU architecture［J］.Proceedings of the VLDB Endowment，2008，1（2）：1313－1324.

［10］ KIM C，PARK J，SATISH N，et al.CloudRAMSort：fast and efficient large－scale distributed RAM sort on shared－nothing cluster［C］／／Proceedings of the 2012 ACM SIGMOD International Conference on Management of Data.ACM，2012：841－850.

［11］ ALBUTIU M C，KEMPER A，NEUMANN T.Massively parallel sort－merge joins in main memory multi－core database systems［J］.Proceedings of the VLDB Endowment，2012，5（10）：1064－1075.

［12］ ZAHARIA M，CHOWDHURY M，DAS T，et al.Resilient distributed datasets：A fault－tolerant abstraction for in－memory cluster computing［C］／／Proceedings of the 9th USENIX conference on Networked Systems Design and Implementation.USENIX Association，2012：2－2.

［13］ XIN R S，ROSEN J，ZAHARIA M，et al.Shark：SQL and rich analytics at scale［C］／／Proceedings of the 2013 international conference on Management of data.ACM，2013，13－24.