李　成　許胤龍　郭　帆　吳　思

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

?

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

李成許胤龍郭帆吳思

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

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

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

0引言

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

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

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

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

1內(nèi)存Join算法

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

1.1Radix Join

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

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

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

1.2并行Radix Join

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

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

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

MapReduce的動(dòng)態(tài)調(diào)度機(jī)制和容錯(cuò)機(jī)制，可以很好地解決傳統(tǒng)并行內(nèi)存連接算法的問(wèn)題。根據(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)通過(guò)Map函數(shù)產(chǎn)生中間數(shù)據(jù)。為了區(qū)分R表與S表的表項(xiàng)，通過(guò)使用添加標(biāo)簽的方式，產(chǎn)生對(duì)應(yīng)的鍵值對(duì)。標(biāo)記的鍵值對(duì)以進(jìn)行連接的項(xiàng)作為鍵。Map函數(shù)的輸出將按鍵的值進(jìn)行排序。所有的具有相同鍵的數(shù)據(jù)會(huì)被劃歸為一組，交由一個(gè)Reducer處理。執(zhí)行過(guò)程如算法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)簽鍵值對(duì)

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算法問(wèn)題的同時(shí)，也帶來(lái)了新的挑戰(zhàn)。由于在鍵值對(duì)生成過(guò)程中，需要以添加標(biāo)簽的方式，讓Reducer區(qū)分是R表還是S表的表項(xiàng)，使得添加的標(biāo)簽處理太多。同時(shí)，標(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í)行過(guò)程，并提出了內(nèi)存Join算法在使用MapReduce框架時(shí)帶來(lái)的挑戰(zhàn)。

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

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ù)塊對(duì)應(yīng)合并

13. add T to ArrayList_R

14. else

15. add T to ArrayList_S

16. /*使用Radix hash對(duì)兩個(gè)數(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ù)。各個(gè)Map任務(wù)通過(guò)使用添加標(biāo)簽的方式，對(duì)數(shù)據(jù)表中的每個(gè)進(jìn)行連接的表項(xiàng)進(jìn)行處理，并產(chǎn)生對(duì)應(yīng)的鍵值對(duì)。在鍵值對(duì)生成過(guò)程中，鍵值對(duì)將按鍵值進(jìn)行插入排序。由于對(duì)單個(gè)鍵值對(duì)添加標(biāo)簽，使得添加的標(biāo)簽處理太多，并且，中間需要排序的數(shù)據(jù)規(guī)模太大，嚴(yán)重影響算法的執(zhí)行性能，本文將對(duì)此進(jìn)行改進(jìn)。

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

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

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

2.2.2Reduce階段優(yōu)化

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

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

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

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

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

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

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

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

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

本文通過(guò)機(jī)器和任務(wù)參數(shù)配置來(lái)模擬不同場(chǎng)景下的內(nèi)存Join算法。使用不同大小的數(shù)據(jù)輸入表進(jìn)行不同數(shù)據(jù)規(guī)模下的性能對(duì)比。實(shí)驗(yàn)結(jié)果顯示，本文的算法性能在所有測(cè)試情況下都優(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)將評(píng)估本文算法隨核數(shù)變化的可擴(kuò)展性。通過(guò)固定R表與S表規(guī)模(共1 GB條記錄)，改變處理數(shù)據(jù)使用的核數(shù)，評(píng)估算法對(duì)于處理器核數(shù)變化的可擴(kuò)展性方面的性能。

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

圖6展現(xiàn)了在固定核數(shù)的情況下，本文算法與已有相關(guān)算法在CMP環(huán)境下隨著數(shù)據(jù)規(guī)模輸入的增大，執(zhí)行時(shí)間的變化。由圖可知，在各個(gè)數(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ī)模的增大，各個(gè)算法的執(zhí)行時(shí)間都有顯著的增大。并且隨著數(shù)據(jù)規(guī)模的增大，改進(jìn)算法相對(duì)標(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動(dòng)態(tài)調(diào)度更能突出其優(yōu)勢(shì)，而樸素的MapReduce Join算法因?yàn)榇罅康靥砑訕?biāo)簽操作以及中間數(shù)據(jù)排序操作花費(fèi)了太多時(shí)間。實(shí)驗(yàn)結(jié)果表明，將原分布式環(huán)境下MapReduce編程模型簡(jiǎn)單搬到內(nèi)存共享環(huán)境下并不能取得突出的性能表現(xiàn)，需要根據(jù)環(huán)境特征重新設(shè)計(jì)算法，才能取得良好的性能。

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

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

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

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

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

4結(jié)語(yǔ)

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

參考文獻(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ù)庫(kù)優(yōu)化查詢。許胤龍，教授。郭帆，碩士生。吳思，博士生。

中圖分類號(hào)TP3

文獻(xiàn)標(biāo)識(shí)碼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