LCDJ：面向內(nèi)存集群計(jì)算的局部感知連接算法

張 磊， 周敏奇， 王 立

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

1 引 言

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

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

分布式內(nèi)存計(jì)算環(huán)境下的哈希連接算法涉及混洗［4］和連接等多個(gè)操作算子.首先，從分布式文件系統(tǒng)中分別讀取待連接的兩張表；然后經(jīng)過(guò)數(shù)據(jù)重分區(qū)將兩張表中連接屬性哈希值相同的記錄傳輸?shù)较嗤?jié)點(diǎn)；最后兩張表屬性哈希值相同的數(shù)據(jù)在單節(jié)內(nèi)部做連接.在此過(guò)程中，混洗涉及網(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)雖然對(duì)混洗過(guò)程都有所優(yōu)化，但是它們并沒(méi)有對(duì)兩張表連接屬性上的數(shù)據(jù)分布做精確統(tǒng)計(jì).本文所提出的LCDJ算法在優(yōu)化網(wǎng)絡(luò)傳輸上起到了較好效果.在既定的數(shù)據(jù)分布環(huán)境下，LCDJ算法精確統(tǒng)計(jì)待連接表在其連接屬性上的屬性值分布，進(jìn)而構(gòu)建以處理時(shí)間為開(kāi)銷的代價(jià)模型，通過(guò)選擇節(jié)點(diǎn)生成優(yōu)化連接方案，大幅減少了時(shí)間開(kāi)銷.本文貢獻(xiàn)為：（1）建立了以處理時(shí)間為開(kāi)銷，考慮網(wǎng)絡(luò)傳輸代價(jià)、并行度和計(jì)算負(fù)載均衡的代價(jià)模型；（2）相對(duì)于枚舉方法選擇連接節(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é)將給出問(wèn)題定義；第四節(jié)詳細(xì)論述LCDJ算法，且建立代價(jià)模型分析算法各個(gè)階段的性能消耗；第五節(jié)將介紹基于原型系統(tǒng)的連接算法實(shí)驗(yàn)和相關(guān)性能分析；第六節(jié)總結(jié)全文.

2 相關(guān)工作

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

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

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

3 問(wèn)題定義

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

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

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

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

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

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

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

4.1 統(tǒng)計(jì)分布

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

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

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

4.2 篩選優(yōu)化

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

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

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

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

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

由此可得出未被選擇作為連接節(jié)點(diǎn)時(shí)，需要傳輸?shù)臄?shù)據(jù)量是此節(jié)點(diǎn)上所有的相關(guān)數(shù)據(jù)，例如對(duì)ri來(lái)說(shuō)，其所在節(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)個(gè)數(shù)越多，平均網(wǎng)絡(luò)傳輸量就越小.但總體時(shí)間開(kāi)銷并非隨節(jié)點(diǎn)個(gè)數(shù)增多而減小，因?yàn)橄到y(tǒng)需要調(diào)度的任務(wù)個(gè)數(shù)也會(huì)增大，所以并行度因素同樣影響性能，下面我們將并行度因素考慮進(jìn)模型中.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

4.3 分布式哈希連接

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

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

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

流水線式exchange exchange會(huì)將兩張表相同哈希段內(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)存之間的傳輸，同時(shí)exchange還被實(shí)現(xiàn)為防止數(shù)據(jù)傾斜，使網(wǎng)絡(luò)帶寬穩(wěn)定維持在最大利用狀態(tài).實(shí)驗(yàn)證明，exchange如果將map端的數(shù)據(jù)持久化到磁盤(pán)后傳輸，遠(yuǎn)不及直接傳輸?shù)乃俣?，后者是前者速度?－3倍，對(duì)于作為瓶頸的網(wǎng)絡(luò)帶寬來(lái)說(shuō)，exchange操作使網(wǎng)絡(luò)帶寬達(dá)到極限，也與我們前面的代價(jià)模型相符.

單節(jié)點(diǎn)無(wú)分區(qū)哈希連接在單節(jié)點(diǎn)內(nèi)部采用無(wú)分區(qū)哈希連接算法［7］，雖然radix連接［6］是能夠針對(duì)硬件配置優(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ī)連接算法，故在此采用無(wú)分區(qū)并行哈希連接算法，build階段多個(gè)線程（圖4中為四線程）遍歷R表建立共享哈希表，probe階段遍歷S表匹配哈希表中的記錄.

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

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

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

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

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

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

5.1 各階段分析

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

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

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

5.1.2 篩選時(shí)間

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

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

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

圖6 篩選時(shí)間比較Fig.6 Comparison of select－times

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

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

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

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

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

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

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

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

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

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

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

圖9 與Hive、Shark對(duì)比Fig.9 Comparison with Hive and Shark

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

6 總 結(jié)

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