基于MapReduce的內(nèi)存并行Join算法研究

李　成　許胤龍　郭　帆　吳　思

(中國科學(xué)技術(shù)大學(xué)計算機(jī)科學(xué)與技術(shù)學(xué)院　安徽 合肥 230027) (安徽省高性能計算重點(diǎn)實(shí)驗(yàn)室　安徽 合肥 230027)

?

基于MapReduce的內(nèi)存并行Join算法研究

李成許胤龍郭帆吳思

(中國科學(xué)技術(shù)大學(xué)計算機(jī)科學(xué)與技術(shù)學(xué)院安徽 合肥 230027) (安徽省高性能計算重點(diǎn)實(shí)驗(yàn)室安徽 合肥 230027)

摘要傳統(tǒng)的并行Join算法缺少必要的容錯能力，且數(shù)據(jù)劃分不均往往導(dǎo)致單個線程的阻塞成為整個任務(wù)執(zhí)行的瓶頸。針對以上問題，分析內(nèi)存連接的各個階段對Join算法性能的影響，提出一種可利用MapReduce的動態(tài)機(jī)制，避免了傳統(tǒng)并行連接算法的數(shù)據(jù)任務(wù)分派不均和容錯問題。算法使用MapReduce編程框架，并通過封裝分塊標(biāo)記減少M(fèi)apReduce Join執(zhí)行過程中標(biāo)記和排序的計算開銷，使算法性能顯著提高。實(shí)驗(yàn)結(jié)果表明，該算法在共享內(nèi)存體系結(jié)構(gòu)下，性能上相比已有算法有顯著改進(jìn)。

關(guān)鍵詞內(nèi)存連接數(shù)據(jù)封裝MapReduce

0引言

當(dāng)前，隨著大數(shù)據(jù)時代的來臨，MapReduce由于其具有良好的可擴(kuò)展性和容錯性，已經(jīng)被廣泛應(yīng)用于面向數(shù)據(jù)處理的應(yīng)用中。MapReduce最初是由谷歌工程師Dean等人在2004年推出[1]，其最初的設(shè)計目的是處理公司大規(guī)模的網(wǎng)絡(luò)日志數(shù)據(jù)訪問。MapReduce編程模式通過提供一種簡單的編程接口實(shí)現(xiàn)在普通機(jī)器上的并行和大規(guī)模分布式計算，并能將并行數(shù)據(jù)處理中容錯、負(fù)載均衡和數(shù)據(jù)分布的復(fù)雜細(xì)節(jié)隱藏起來，自動完成[2]。

Join 算法是進(jìn)行兩個或者多個數(shù)據(jù)集聚集連接的操作，是面向數(shù)據(jù)處理應(yīng)用的常用算法。優(yōu)化Join執(zhí)行的效率，可以有效提升數(shù)據(jù)分析任務(wù)的性能。Join在分布式環(huán)境下使用MapReduce并行化的研究成果相對豐富[3-5]，但針對內(nèi)存共享環(huán)境下，使用MapReduce對Join算法進(jìn)行并行化的研究卻仍十分少見[10]。然而，隨著多核處理器的普及和內(nèi)存數(shù)據(jù)庫的流行，研究MapReduce在共享內(nèi)存環(huán)境下實(shí)現(xiàn)內(nèi)存Join算法的并行具有重要意義。

傳統(tǒng)的內(nèi)存Join算法，研究內(nèi)容主要集中在均攤并行處理數(shù)據(jù)的多個線程的任務(wù)和優(yōu)化cache訪問兩個方向。其中，一種流行的最新技術(shù)是使用Radix-Cluster Hash Join[9]。但是，針對內(nèi)存多核體系結(jié)構(gòu)，當(dāng)有多個線程并行處理已經(jīng)存放到內(nèi)存的數(shù)據(jù)時，直接使用以上策略并不能充分發(fā)揮其性能。傳統(tǒng)并行方式通過多個線程并行執(zhí)行任務(wù)子集，將帶來單個線程任務(wù)過大時會成為整個查詢事務(wù)的瓶頸，特別是單個線程查詢失敗將導(dǎo)致整個查詢失敗。

本文提出通過引入MapReduce的機(jī)制解決傳統(tǒng)并行Join算法單個線程成為算法瓶頸或者導(dǎo)致整個任務(wù)失敗的問題。在標(biāo)準(zhǔn)MapReduce Join算法的基礎(chǔ)上，結(jié)合多核體系架構(gòu)的特性，提出了基于MapReduce的 Radix Join優(yōu)化算法。在該算法中考慮了cache 命中率和MapReduce執(zhí)行過程中數(shù)據(jù)分片規(guī)模對算法的影響，在減少中間結(jié)果規(guī)模的同時，保證算法具有良好的cache 敏感特性。在CMP和SMP環(huán)境下的實(shí)驗(yàn)結(jié)果表明，該算法無論是對比傳統(tǒng)內(nèi)存共享并行Join算法還是常用的標(biāo)準(zhǔn)MapReduce Join算法，性能均具有較大提升。

1內(nèi)存Join算法

內(nèi)存Join算法優(yōu)化的研究眾多[6-8]，并提出了多種針對不同情形下的優(yōu)秀算法，其中，Radix Join 便是針對等值Join的突出代表。下面將介紹該算法串行及其并行算法的執(zhí)行過程。

1.1Radix Join

Balkesen等[9]證實(shí)了當(dāng)哈希表大于cache的大小時，幾乎每個訪問都導(dǎo)致一次cache訪問缺失。因此，切分哈希表，使每個哈希表的大小能夠小于cache的大小，可以提升系統(tǒng)性能。Albutiu等[7]借鑒該思想，通過考慮傳輸后備緩沖器(TLB)對性能的影響，提出了多次劃分的算法思想。現(xiàn)在該思想已經(jīng)成為Radix Join算法的標(biāo)準(zhǔn)組成。

完整的Radix Join說明如圖1所示。兩個輸入都是通過使用兩次Radix數(shù)據(jù)劃分的方式劃分到合適的大小。每個ri由基于哈希劃分輸入R得到， ri會根據(jù)哈希函數(shù)進(jìn)行第二次劃分。所有的sj劃分的分區(qū)會被遍歷并與ri所劃分成的哈希子表中的表項(xiàng)進(jìn)行連接匹配。在Radix Join中，為了取得良好的cache特性，避免一次過多的數(shù)據(jù)片劃分產(chǎn)生，兩個輸入表都需要經(jīng)過多段的數(shù)據(jù)劃分處理。

圖1　Radix Join執(zhí)行過程[7]

1.2并行Radix Join

對于通過將劃分過程中產(chǎn)生的數(shù)據(jù)子集由相互獨(dú)立的多個線程并行執(zhí)行，串行的Radix Join 算法可以實(shí)現(xiàn)算法的并行化[8]。在第一階段，由單獨(dú)的線程劃分?jǐn)?shù)據(jù)，并對于每個線程將會產(chǎn)生自己專用的部分?jǐn)?shù)據(jù)的數(shù)據(jù)區(qū)域[7]。在第一步數(shù)據(jù)劃分完成以后，各個任務(wù)已經(jīng)具有足夠的獨(dú)立性，可以很好地并行完成各自的工作。線程工作的任務(wù)分發(fā)通過任務(wù)隊(duì)列實(shí)現(xiàn)。通過以上方法，對于一個p核系統(tǒng)，該算法的時間復(fù)雜度可以期望為單核的1/p。

上述算法在數(shù)據(jù)均勻分布時具有較好的并行特性。但是，算法在進(jìn)行數(shù)據(jù)劃分時，很可能導(dǎo)致劃分?jǐn)?shù)據(jù)的失衡，從而導(dǎo)致在并行執(zhí)行階段中，數(shù)據(jù)處理時間最長的線程成為整個任務(wù)的瓶頸。更重要的是，在并行執(zhí)行階段，一旦某個線程處理出了問題，將會導(dǎo)致整個查詢?nèi)蝿?wù)的失敗。

2MapReduce Join算法及其優(yōu)化

MapReduce的動態(tài)調(diào)度機(jī)制和容錯機(jī)制，可以很好地解決傳統(tǒng)并行內(nèi)存連接算法的問題。根據(jù)MapReduce的特征，MapReduce Join算法可以有兩類實(shí)現(xiàn)：Map-side Join和Reduce-side Join[11]。由于Map-side Join算法要求數(shù)據(jù)是有序的[11]，因此，本文只關(guān)注適用范圍更廣的Reduce-side Join。

2.1樸素的MapReduce Join算法

如圖2所示，Reduce-side Join 算法將輸入數(shù)據(jù)的表項(xiàng)通過Map函數(shù)產(chǎn)生中間數(shù)據(jù)。為了區(qū)分R表與S表的表項(xiàng)，通過使用添加標(biāo)簽的方式，產(chǎn)生對應(yīng)的鍵值對。標(biāo)記的鍵值對以進(jìn)行連接的項(xiàng)作為鍵。Map函數(shù)的輸出將按鍵的值進(jìn)行排序。所有的具有相同鍵的數(shù)據(jù)會被劃歸為一組，交由一個Reducer處理。執(zhí)行過程如算法1所描述。

圖2　Reduce-side Join 數(shù)據(jù)流[11]

算法1樸素的MapReduce Join算法

Require: Input relations R and S for Join operations

1.Map(Key k, Value v)

//map 階段

2. if(v comes from R)

//標(biāo)記R表和S表表項(xiàng)

3. tag=1 and Join_key=v.a

4. else

5. tag=2 and Join_key=v.b

6. Output.collect( Join_key, T+tag)

//輸出帶標(biāo)簽鍵值對

7.end Map

8.Reduce( key key,List values)

//reduce 階段

9. for each v in values

10. if(v.tag=1)

//根據(jù)標(biāo)記將數(shù)據(jù)加入相應(yīng)數(shù)據(jù)集

11. add v to ArrayList_R

12. else

13. add v to ArrayList_S

14. for each val1 in ArrayList_R

15. for each val2 in ArrayList_S

16. result=val1 Join val2

//執(zhí)行join

17. collect(key, result)

18.end reduce

MapReduce的引入，在解決傳統(tǒng)并行Radix Join算法問題的同時，也帶來了新的挑戰(zhàn)。由于在鍵值對生成過程中，需要以添加標(biāo)簽的方式，讓Reducer區(qū)分是R表還是S表的表項(xiàng)，使得添加的標(biāo)簽處理太多。同時，標(biāo)準(zhǔn)MapReduce編程框架要求中間結(jié)果將按鍵值進(jìn)行排序，需要排序的數(shù)據(jù)規(guī)模太大，將嚴(yán)重影響算法的執(zhí)行性能。另外，為了保持Radix的cache特性，數(shù)據(jù)的最終劃分也需要合理的選擇。因此，本文提出了一種新的改進(jìn)方法。

2.2MapReduce Join算法優(yōu)化

在上文中介紹了樸素的MapReduce Join算法的實(shí)現(xiàn)及其執(zhí)行過程，并提出了內(nèi)存Join算法在使用MapReduce框架時帶來的挑戰(zhàn)。

在多核體系架構(gòu)下，樸素的MapReduce的算法設(shè)計暴露出其弊端。內(nèi)存和數(shù)據(jù)的訪問執(zhí)行是通過如圖3所示的體系完成的。基于該體系架構(gòu)內(nèi)的通信代價幾乎是可以忽略的，但是MapReduce標(biāo)準(zhǔn)執(zhí)行中的標(biāo)記及排序操作將成為算法的主要開銷。分布式MapReduce環(huán)境下的Join算法優(yōu)化，大多關(guān)注于網(wǎng)絡(luò)通信代價的優(yōu)化，缺少對于內(nèi)存共享環(huán)境下，結(jié)合計算機(jī)多核特性的Join算法的深入研究。基于以上原因，本文將分析Join 算法的執(zhí)行過程，從而提出并實(shí)現(xiàn)適合內(nèi)存共享環(huán)境下的MapReduce Join算法。針對如圖4所示的數(shù)據(jù)流，根據(jù)算法2的執(zhí)行過程，分別對Map和Reduce兩個階段對算法進(jìn)行優(yōu)化。

圖3　三層cache的多核體系結(jié)構(gòu)

圖4　內(nèi)存MapReduce Join的數(shù)據(jù)流

算法2改進(jìn)的MapReduce Join算法

Require: Input relations R and S for Join operations

1.Map(Key k, Values Vs)

//map 階段

2. /*使用Radix hash進(jìn)行第一次劃分后封裝*/

3. Use the hash1 to split the Vs into blocks Ts

4. if(T comes from R)

//標(biāo)記R表和S表數(shù)據(jù)塊

5. tag=1 Join_key=hash1(T.a)

6. else

7. tag=2 Join_key=hash1(T.b)

8. Output.collect( Join_key, T+tag)

//輸出帶標(biāo)簽鍵值對

9.end map

10.reduce (Key k’, List blocks)

//Reduce 階段

11. for each T in blocks

12. if(T.tag=1)

//根據(jù)標(biāo)記將數(shù)據(jù)塊對應(yīng)合并

13. add T to ArrayList_R

14. else

15. add T to ArrayList_S

16. /*使用Radix hash對兩個數(shù)據(jù)集進(jìn)行第二次劃分*/

17. split ArrayList_R with hasp into ArrayList_R’

18. split ArrayList_R with hasp into ArrayList_S’

19. for each val1 in ArrayList_R’

20. for each val2 in ArrayList_S’

21. do result=val1 Join val2

//執(zhí)行join

22. collect(key, result)

23.end reduce

2.2.1Map階段優(yōu)化

在數(shù)據(jù)劃分之后，相互獨(dú)立的Map任務(wù)并行處理分配給自己的數(shù)據(jù)。各個Map任務(wù)通過使用添加標(biāo)簽的方式，對數(shù)據(jù)表中的每個進(jìn)行連接的表項(xiàng)進(jìn)行處理，并產(chǎn)生對應(yīng)的鍵值對。在鍵值對生成過程中，鍵值對將按鍵值進(jìn)行插入排序。由于對單個鍵值對添加標(biāo)簽，使得添加的標(biāo)簽處理太多，并且，中間需要排序的數(shù)據(jù)規(guī)模太大，嚴(yán)重影響算法的執(zhí)行性能，本文將對此進(jìn)行改進(jìn)。

針對Map階段的優(yōu)化，本文使用封裝標(biāo)記法減少算法執(zhí)行過程中的計算開銷。因?yàn)樵贛ap階段需要將所有數(shù)據(jù)是來自R表還是S表進(jìn)行標(biāo)記，針對每個表項(xiàng)進(jìn)行標(biāo)記執(zhí)行代價太高。由于面向的是等值連接，可以使用哈希的方式進(jìn)行初次的數(shù)據(jù)劃分。如算法2的第3～5行描述，劃分后的數(shù)據(jù)塊封裝成一個整體，將該數(shù)據(jù)塊的哈希值作為鍵，包含有封裝數(shù)據(jù)地址的數(shù)據(jù)結(jié)構(gòu)作為值，生成鍵值對。通過這種封裝的方式將每個鍵值對的標(biāo)記將是對每個數(shù)據(jù)集進(jìn)行整體標(biāo)記，減少了大量標(biāo)記操作。由于中間的結(jié)果數(shù)據(jù)需要進(jìn)行排序，適當(dāng)?shù)姆庋b同時也減少了中間結(jié)果需要排序的數(shù)量。

為了具備良好的cache特性，算法利用Radix Hash函數(shù)對數(shù)據(jù)進(jìn)行劃分。但是對于大規(guī)模數(shù)據(jù)而言(GB級以上)，如果Radix哈希的劃分分片太少，將不能充分地發(fā)揮MapReduce動態(tài)調(diào)度的優(yōu)勢。可能導(dǎo)致數(shù)據(jù)分配不均衡，使得單個Reducer任務(wù)成為瓶頸而無法充分發(fā)揮并行能力。為了解決此問題，本文通過多次劃分的方式解決。在Map階段進(jìn)行適當(dāng)規(guī)模的數(shù)據(jù)劃分，初次劃分的規(guī)模，經(jīng)過實(shí)驗(yàn)驗(yàn)證，如圖5所示。每次數(shù)據(jù)劃分使用一個字節(jié)進(jìn)行劃分，這樣產(chǎn)生256個分組，將得到性能最接近最優(yōu)的劃分。以字節(jié)的方式劃分，可以通過整個字節(jié)截取的方式，在減少了中間數(shù)據(jù)存儲開銷的同時，也使得一次可以有更多的數(shù)據(jù)放入cache中，提高cache命中率。對于任務(wù)可能的分布不均，將對數(shù)據(jù)規(guī)模超過參數(shù)限制(本文中為標(biāo)準(zhǔn)值8倍)的數(shù)據(jù)塊進(jìn)行多一次的劃分。

圖5　Radix哈希分塊使用位數(shù)變化對性能的影響

2.2.2Reduce階段優(yōu)化

本文不僅關(guān)注Map執(zhí)行階段的優(yōu)化，還關(guān)注Reduce階段的優(yōu)化。Reducers等待所有的Map任務(wù)的返回結(jié)果。中間結(jié)果中針對每個hash值的數(shù)據(jù)，都會調(diào)用一次Reduce 任務(wù)。

每個Reduce任務(wù)負(fù)責(zé)處理多個Reduce數(shù)據(jù)塊(如圖4所示)。每個Map任務(wù)處理部分的數(shù)據(jù)，Reducer歸并所有部分的數(shù)據(jù)放在合適的緩存中。Reducer需要計算所有具有相同鍵值的中間值，并輸出數(shù)據(jù)最終的處理結(jié)果。正如本文前面介紹的Radix Join所示，需要設(shè)計合理的調(diào)度策略來優(yōu)化數(shù)據(jù)的訪問性能。

如Map階段所描述，為了減少中間結(jié)果排序時間等，將Map階段劃分的數(shù)據(jù)塊數(shù)控制在一個較小的規(guī)模。這使得每個Reduce任務(wù)需要處理的數(shù)據(jù)規(guī)模過大。本文通過對數(shù)據(jù)進(jìn)行二次劃分，并將數(shù)據(jù)切分到可以容納至最低cache層，進(jìn)行對多個數(shù)據(jù)集的并行執(zhí)行。

為了最優(yōu)化內(nèi)存駐留的算法，選擇一個簡單并且合適的內(nèi)存訪問體系模型將非常重要。盡管本文選用的多核結(jié)構(gòu)的內(nèi)存體系(如圖3所示)，相對于真實(shí)的內(nèi)存體系稍微簡單，但其是不同平臺共有的基本框架[12]。本文的算法是cache敏感的，其優(yōu)化主要針對第一層cache，也就是L1層的cache優(yōu)化。因?yàn)閷τ诓煌钠脚_，高層的cache多有變化，而上層的cache訪問往往不會影響最低層的cache訪問性能，這使得本文的研究具有普遍的適用性。本文將進(jìn)行連接的最終數(shù)據(jù)規(guī)模劃分到不能超過L1層的cache大小，這樣既避免了最低層的cache震蕩，還可以適度地優(yōu)化高層的cache命中率。

3實(shí)驗(yàn)及結(jié)果分析

為了評價本文的算法，我們在不同環(huán)境下實(shí)現(xiàn)了文中所提到的樸素算法和改進(jìn)算法，并對比已有的傳統(tǒng)Radix Join算法并行方法。實(shí)驗(yàn)運(yùn)行在一個具有16核的 Intel Xeon SMP 硬件系統(tǒng)上，操作系統(tǒng)為64位版本的CentOS Linux 6.4。

3.1實(shí)驗(yàn)數(shù)據(jù)

為了更好地對比，本文采用的數(shù)據(jù)集與文獻(xiàn)[8]中使用的數(shù)據(jù)集相似。輸入的數(shù)據(jù)全部是整數(shù)，特別是已有算法的假設(shè)數(shù)據(jù)場景是面向列式存儲的，所以我們將數(shù)據(jù)表執(zhí)行定義為每行僅是鍵值對的表項(xiàng)，每條記錄長度為16 B，鍵和值的數(shù)據(jù)類型是8個字節(jié)的長整數(shù)。實(shí)驗(yàn)所測試的數(shù)據(jù)R表與S表數(shù)據(jù)規(guī)模相等，由于硬件條件的限制,為了防止數(shù)據(jù)刷到虛擬內(nèi)存,影響算法性能的測量, 最大數(shù)據(jù)集為兩個表各5億條記錄,共16 GB。

3.2實(shí)驗(yàn)設(shè)置

本文通過機(jī)器和任務(wù)參數(shù)配置來模擬不同場景下的內(nèi)存Join算法。使用不同大小的數(shù)據(jù)輸入表進(jìn)行不同數(shù)據(jù)規(guī)模下的性能對比。實(shí)驗(yàn)結(jié)果顯示，本文的算法性能在所有測試情況下都優(yōu)于其他的算法。

在前兩組實(shí)驗(yàn)中，依次在CMP(8核)和SMP(16核)環(huán)境下固定核數(shù)，使用數(shù)據(jù)規(guī)模為1、2、4、8、16 GB的不同規(guī)模輸入的數(shù)據(jù)集進(jìn)行實(shí)驗(yàn)。最后的一組實(shí)驗(yàn)將評估本文算法隨核數(shù)變化的可擴(kuò)展性。通過固定R表與S表規(guī)模(共1 GB條記錄)，改變處理數(shù)據(jù)使用的核數(shù)，評估算法對于處理器核數(shù)變化的可擴(kuò)展性方面的性能。

3.3實(shí)驗(yàn)結(jié)果及分析

圖6展現(xiàn)了在固定核數(shù)的情況下，本文算法與已有相關(guān)算法在CMP環(huán)境下隨著數(shù)據(jù)規(guī)模輸入的增大，執(zhí)行時間的變化。由圖可知，在各個數(shù)據(jù)集下，改進(jìn)的MapReduce join 算法均優(yōu)于標(biāo)準(zhǔn)的Radix Join 并行化實(shí)現(xiàn)，而樸素的MapReduce Join算法性能最差。隨著實(shí)驗(yàn)數(shù)據(jù)規(guī)模的增大，各個算法的執(zhí)行時間都有顯著的增大。并且隨著數(shù)據(jù)規(guī)模的增大，改進(jìn)算法相對標(biāo)準(zhǔn)并行Radix Join和樸素MapReduce Join性能提升更加明顯，由1 GB數(shù)據(jù)規(guī)模的性能分別提升了28.1%和77.3%，到16 GB數(shù)據(jù)規(guī)模的性能分別提升了46.7%和77.9%。這是因?yàn)殡S著數(shù)據(jù)規(guī)模的增加，MapReduce動態(tài)調(diào)度更能突出其優(yōu)勢，而樸素的MapReduce Join算法因?yàn)榇罅康靥砑訕?biāo)簽操作以及中間數(shù)據(jù)排序操作花費(fèi)了太多時間。實(shí)驗(yàn)結(jié)果表明，將原分布式環(huán)境下MapReduce編程模型簡單搬到內(nèi)存共享環(huán)境下并不能取得突出的性能表現(xiàn)，需要根據(jù)環(huán)境特征重新設(shè)計算法，才能取得良好的性能。

圖6　Join算法在8核CMP系統(tǒng)上的性能對比

圖7展示了SMP環(huán)境下(16核)處理與CMP(8核)環(huán)境下相同數(shù)據(jù)集的各個算法的處理時間。由圖可知，各算法在16核環(huán)境下執(zhí)行時間都有不同程度的減少，但在SMP環(huán)境下，改進(jìn)MapReduce Join算法相對其他兩種算法的性能仍然有很大提升。以實(shí)驗(yàn)數(shù)據(jù)1 GB的第一組實(shí)驗(yàn)為例，改進(jìn)的MapReduce Join的執(zhí)行時間由CMP環(huán)境下的0.983 s下降到0.6196 s，相對于標(biāo)準(zhǔn)并行Radix Join和樸素MapReduce Join性能分別提升了26.9%和76.6%。相對CMP環(huán)境下提升雖然略有下降，但基本上和CMP(8核)環(huán)境下取得了一致的結(jié)果。

圖7　Join算法在16核SMP系統(tǒng)上的性能對比

圖8展示了對于同一數(shù)據(jù)集，各個算法隨著計算核數(shù)變化的執(zhí)行時間變化。該擴(kuò)展性測試顯示，各個算法隨著核數(shù)的增加，執(zhí)行時間逐漸減少，而本文所提出算法的執(zhí)行時間隨核數(shù)增加而下降最為迅速。因?yàn)殡S著實(shí)際使用核數(shù)的增加，將會有更多的線程同時在共享數(shù)據(jù)的情況下進(jìn)行數(shù)據(jù)處理，使得每個Map任務(wù)或者Reduce任務(wù)處理的數(shù)據(jù)規(guī)模減少。在單核環(huán)境下，雖然算法的執(zhí)行時間稍遜色于標(biāo)準(zhǔn)并行Radix Join，但當(dāng)核數(shù)增多后，由于并行處理數(shù)據(jù)劃分等原因，改進(jìn)的MapReduce Join算法表現(xiàn)的性能開始超過標(biāo)準(zhǔn)并行Radix Join。并且最終隨著核數(shù)的增加，算法性能的提升呈現(xiàn)保持的趨勢。圖8展示的對于擴(kuò)展性能的測試結(jié)果驗(yàn)證了改進(jìn)后的MapReduce Join算法不僅具有高效性，還具有良好的可擴(kuò)展性。

圖8　Join算法擴(kuò)展性測試

4結(jié)語

本文提出一種新的內(nèi)存Join算法，且該算法在多核共享內(nèi)存體系結(jié)構(gòu)下可以取得高效性能。該算法借助MapReduce編程框架，并利用Radix算法的特性，在標(biāo)準(zhǔn)實(shí)現(xiàn)上加以改進(jìn)，解決了傳統(tǒng)并行Join算法單個線程阻塞成為整個任務(wù)瓶頸以及缺少容錯性的問題。通過在Map階段劃分后封裝減少中間結(jié)果數(shù)據(jù)規(guī)模，解決了因引入MapReduce方式帶來中間結(jié)果標(biāo)記和排序開銷過大的問題，使得算法在具有了MapReduce良好容錯性的同時，具有高效性。新的MapReduce Join算法在多核內(nèi)存共享環(huán)境下，相對于原有算法，在計算性能和良好的可擴(kuò)展性方面均具有突出的優(yōu)勢。

參考文獻(xiàn)

[1] Dean J,Ghemawat S.MapReduce:simplified data processing on large clusters[J].Communications of the ACM,2008,51(1):107-113.

[2] Ranger C,Raghuraman R,Penmetsa A,et al.Evaluating MapReduce for multi-core and multiprocessor systems[C]//High Performance Computer Architecture,2007.HPCA 2007.IEEE 13th International Symposium on.IEEE,2007:13-24.

[3] Afrati F N,Ullman J D.Optimizing Joins in a Map-Reduce environment[C]//Proceedings of the 13th International Conference on Extending Database Technology.ACM,2010:99-110.

[4] Blanas S,Patel J M,Ercegovac V,et al.A comparison of Join algorithms for log processing in MapReduce[C]//Proceedings of the 2010 ACM SIGMOD International Conference on Management of data.ACM,2010:975-986.

[5] Afrati F N,Ullman J D.Optimizing multiway Joins in a Map-Reduce environment[J].Knowledge and Data Engineering,IEEE Transactions on,2011,23(9):1282-1298.

[6] 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.

[7] 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.

[8] Balkesen C,Teubner J,Alonso G,et al.Main-Memory Hash Joins on Modern Processor Architectures[J].Knowledge and Data Engineering,IEEE Transactions on,2014,26(3):99-113.

[9] 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 IEEE 29th International Conference on.IEEE,2013:362-373.

[10] Jiang W,Ravi V T,Agrawal G.A map-reduce system with an alternate api for multi-core environments[C]//Proc of the 10th Int Conf on Cluster,Cloud,and Grid Computing.IEEE,2010:84-93.

[11] Jadhav V,Aghav J,Dorwani S.Join Algorithms Using MapReduce:A Survey[C]//International Conference on Electrical Engineering and Computer Science,21st.2013.

[12] Boncz P A,Manegold S,Kersten M L.Database architecture optimized for the new bottleneck:Memory access[C]//VLDB,1999,99:54-65.

收稿日期：2015-02-11。李成，碩士生，主研領(lǐng)域：數(shù)據(jù)庫優(yōu)化查詢。許胤龍，教授。郭帆，碩士生。吳思，博士生。

中圖分類號TP3

文獻(xiàn)標(biāo)識碼A

DOI:10.3969/j.issn.1000-386x.2016.07.059

RESEARCH ON MAPREDUCE-BASED IN-MEMORY PARALLEL JOIN ALGORITHM

Li ChengXu YinlongGuo FanWu Si

(SchoolofComputerScienceandTechnology,UniversityofScienceandTechnologyofChina,Hefei230027,Anhui,China) (TheKeyLaboratoryonHighPerformanceComputing,Hefei230027,Anhui,China)

AbstractTraditional parallel Join algorithms lack the necessary fault tolerance capability, and data partitioning inequality often leads to a single thread obstruction which in turn becomes the bottleneck of the whole task execution. In light of the above problem, this paper dissects the influence of each phase of in-memory join on the performance of Join algorithm, and proposes a dynamic mechanism in which the MapReduce is applicable, thus avoids the problems of traditional parallel Join algorithm implementation in unequal data tasks allocation and fault tolerance. The algorithm uses MapReduce programming framework, and reduces the computational cost of tagging and ranking in execution process of MapReduce Join through encapsulating the blocking tags, this makes the performance of the algorithm improve remarkably. Experimental results show that this algorithm has evident improvement in performance for shared-memory architecture.

KeywordsIn-memory JoinData encapsulationMapReduce