Map Reduce模型下的并行線性時間選擇算法研究

王永貴，李鴻緒，宋 曉

（遼寧工程技術(shù)大學(xué)軟件學(xué)院，遼寧葫蘆島125105）

0 引 言

線性時間選擇算法被廣泛應(yīng)用于從若干元素中迅速確定某一個元素［1］。該算法在處理少量的數(shù)據(jù)時是有效的，但是面對大數(shù)據(jù)，算法的執(zhí)行效率不能夠滿足人們的需要。隨著網(wǎng)絡(luò)信息技術(shù)的發(fā)展，人們可以采集并利用的數(shù)據(jù)越來越多，因此如何處理海量數(shù)據(jù)下的線性時間選擇是迫切需要解決的問題。Google提出的Map Reduce［2］并行編程框架在處理海量數(shù)據(jù)問題上有著顯著的優(yōu)勢，該模型具有良好的擴(kuò)展性及容錯性，能夠滿足人們對海量數(shù)據(jù)處理的需要。Hadoop［3］開源系統(tǒng)很好地實(shí)現(xiàn)了MapReduce編程模型，因此在對大數(shù)據(jù)處理的時候Hadoop已經(jīng)成為一種不可或缺的選擇?？梢酝ㄟ^改進(jìn)傳統(tǒng)的線性時間選擇算法，使其適應(yīng)Map Reduce并行編程模型，從而能夠有效地解決海量數(shù)據(jù)下的線性時間選擇問題。

文中對Hadoop項目的研究，提出了利用Hadoop Map Reduce模型來改進(jìn)線性時間選擇算法，使得可以在集群的多個分布式節(jié)點(diǎn)上實(shí)現(xiàn)改進(jìn)后的算法，提高線性時間選擇算法的執(zhí)行效率。

1 Map Reduce計算模型

Map Reduce用于研究和處理大規(guī)模數(shù)據(jù)集的并行計算［4］。Map Reduce是一種可用于數(shù)據(jù)處理的編程模型，MapReduce程序本質(zhì)上是并行運(yùn)行的。在Hadoop集群［5］中，用于執(zhí)行MapReduce并行化任務(wù)的節(jié)點(diǎn)有兩個角色，分別是Job Tracker和Task Tracker，一個Hadoop集群中只能有一個Job Tracker節(jié)點(diǎn)。Hadoop中每個Map Reduce任務(wù)會被初始化成一個Job，每個Job又會自動被分為兩個運(yùn)行階段，分別是Map階段和Reduce階段［6］，這兩個階段是用函數(shù)的形式表示，用戶可以根據(jù)任務(wù)的需求來自行完成函數(shù)的編寫任務(wù)，即完成Map函數(shù)和Reduce函數(shù)的編寫任務(wù)。Map函數(shù)接收一個＜key，value＞鍵值對的輸入，該鍵值對會有Hadoop內(nèi)部系統(tǒng)提供，然后Map函數(shù)會產(chǎn)生中間結(jié)果，也是以＜key，value＞鍵值對的形式輸出，Hadoop會將該中間結(jié)果進(jìn)行適當(dāng)?shù)膮R總并非配給相應(yīng)的Reduce函數(shù)，Reduce函數(shù)會接收到＜key，（list of values）＞的鍵值對輸入，然后對這個鍵值對進(jìn)行處理，并輸出結(jié)果，同樣也是＜key，value＞的形式。

2 Map Reduce任務(wù)執(zhí)行流程

Map Reduce任務(wù)的執(zhí)行流程［7］可根據(jù)用戶的需求分為7個步驟，如圖1所示。各個階段完成的任務(wù)如下:

（1）Input階段:Hadoop會把大到影響它查找效率的文件分割成等長的小數(shù)據(jù)發(fā)送到MapReduce，等長的數(shù)據(jù)稱為輸入分片（input split）［8］，一個分片并不是數(shù)據(jù)本身，而是對數(shù)據(jù)的引用，Hadoop會根據(jù)分片上的信息將數(shù)據(jù)分配給對應(yīng)的Map任務(wù)，數(shù)據(jù)在分配給對應(yīng)的Map任務(wù)之前會被制定成一定的格式。

（2）Map階段:Map任務(wù)會從輸入的分片中讀取記錄，然后調(diào)用用戶自己定義的map函數(shù)，處理后輸出結(jié)果。map函數(shù)是由JobConf.Set MapperClass提交給系統(tǒng)，map函數(shù)處理的是沒有經(jīng)過加工的數(shù)據(jù)，數(shù)據(jù)之間沒有相互依賴的關(guān)系，它會對每條數(shù)據(jù)進(jìn)行相應(yīng)的解析，從中提取key和value值，也就是數(shù)據(jù)的特征。數(shù)據(jù)的處理沒有先后關(guān)系，也不需要和上一個狀態(tài)的信息進(jìn)行交互，它只處理當(dāng)前的數(shù)據(jù)。

（3）Sort階段:對map輸出的中間結(jié)果進(jìn)行相應(yīng)的排序，排序的規(guī)則可以由用戶自己定義，通過設(shè)置Job－Conf.setOutput KeyComparatotClass來實(shí)現(xiàn)，此階段的排序是為了提高M(jìn)ap Reduce的工作效率，縮短任務(wù)的執(zhí)行時間，降低任務(wù)的復(fù)雜度。

（4）Combine階段:在一些情況下，中間結(jié)果的key出現(xiàn)的頻率可能會比較高，導(dǎo)致重復(fù)現(xiàn)象，該階段會將這些重復(fù)的key進(jìn)行合并，再發(fā)送給Reduce任務(wù)，這樣不僅可以減少中間結(jié)果的數(shù)據(jù)重復(fù)，更可以降低網(wǎng)絡(luò)的負(fù)載，提高M(jìn)ap Reduce的執(zhí)行效率。

（5）Partition階段:在Map輸出的中間結(jié)果傳遞給Reduce任務(wù)之前，需要將數(shù)據(jù)進(jìn)行分組，每一組會根據(jù)需求傳遞給不同的Reduce任務(wù)，本階段通過key進(jìn)行分組，從而確定中間數(shù)據(jù)被分發(fā)到哪個Reduce任務(wù)上去執(zhí)行。

（6）Reduce階段:通過上一階段，Map輸出的中間結(jié)果會被分配給對應(yīng)的Reduce任務(wù)，這里需要處理這些中間結(jié)果及中間鍵值相關(guān)集合，需要合并這些中間結(jié)果，合并的任務(wù)由reduce函數(shù)來完成，最后形成較少的鍵值對的集合，并輸出對應(yīng)的結(jié)果到下一環(huán)節(jié)。

（7）Output階段:Reduce的輸出結(jié)果被存放到HDFS文件系統(tǒng)［9］，由JobConf.setOutputFormat來指定輸出的格式，并且輸出的格式和Map任務(wù)的輸入格式是相互對應(yīng)的，因?yàn)镽educe任務(wù)的輸出是可以被其他Map任務(wù)讀入的，從而達(dá)到循環(huán)執(zhí)行Map Reduce任務(wù)的目的。

圖1 MapReduce任務(wù)執(zhí)行流程

3 線性時間選擇算法的Map Reduce實(shí)現(xiàn)

3.1 線性時間選擇算法

線性時間選擇算法是與排序問題類似的元素選擇算法。元素選擇問題的一般概念為:在給定的線性序集中n個元素和一個整數(shù)k，1≤k≤n，要求找出這n個元素中第k小的元素，即如果將這n個元素依其線性排序時，排在第k個位置的元素即為要找的元素。當(dāng)k＝1時，找的就是最小元素；同理，當(dāng)k＝n時，找的就是最大元素；當(dāng)k＝（1＋n）/2時，稱為找中位數(shù)。

在某些特殊情況下，很容易設(shè)計出解選擇問題的線性時間算法。例如，找n個元素的最小元素和最大元素顯然可以在O （n）時間內(nèi)完成。如果k≤n/log n，通過堆排序算法可以在O （n＋k log n ）＝O （n）時間內(nèi)找出第k小元素，當(dāng)k≥n－n/log n時也一樣。

3.2 在Map Reduce框架下的實(shí)現(xiàn)

線性時間選擇問題在Map Reduce框架下實(shí)現(xiàn)的基本思路，為了清晰的表述算法步驟，用n代表元素總個數(shù)，n＞0；m表示總元素被分成的組的個數(shù)，m＞0；k表示要找的第k小元素，k＞0；MIN＿LIST表示最小元素列表，按照升k＞0序排列好的數(shù)據(jù)，初始值為空；t表示最小元素列表中的元素個數(shù)，t≥0；WAITFORSEL＿LIST表示各個分組輸出后的最小值列表。

步驟1 將n個數(shù)據(jù)分成m組，找出每一組元素中的最小元素，并用該最小元素與MIN＿LIST中的數(shù)據(jù)進(jìn)行比較，若最小值列表中存在該元素則找出第二小的元素，同樣將該元素與MIN＿LIST比較，直到該最小元素不存在于MIN＿LIST中，將各個組的最小元素匯總并統(tǒng)一放入WAITFORSEL＿LIST中。

步驟2 找出WAITFORSEL＿LIST列表中的最小元素，加入MIN＿LIST中，MIN＿LIST初始為空。

步驟3 將MIN＿LIST中的元素個數(shù)記為t，若k＝t，則MIN＿LIST中的第t個元素即為要找的第k小元素，算法結(jié)束；若k＞t，則回到步驟一繼續(xù)執(zhí)行算法。

圖2描述了線性時間選擇算法的Map Reduce并行化實(shí)現(xiàn)過程。由于Map Reduce計算模型的需要，數(shù)據(jù)記錄要以行的形式進(jìn)行存儲［10］，使得數(shù)據(jù)能夠按照行來進(jìn)行分片，分片的過程由MapReduce的運(yùn)行環(huán)境來完成，不需要進(jìn)行代碼的編寫。

3.3 map函數(shù)的實(shí)現(xiàn)

map函數(shù)實(shí)現(xiàn)的任務(wù)是計算出本組中的最小元素，且該元素不存在于MIN＿LIST中，否則繼續(xù)尋找最小元素，map函數(shù)的輸入為Map Reduce運(yùn)行環(huán)境分配給該map任務(wù)的元素集合和上一輪更新后的MIN＿LIST，輸入的數(shù)據(jù)記錄為＜key，value＞對的形式，該鍵值對的含義是＜行號，記錄行＞；每個map函數(shù)都要讀入描述MIN＿LIST的文件，map函數(shù)對輸入的所有元素進(jìn)行比較，計算出最小元素，并將該最小元素與MIN＿LIST文件中的元素進(jìn)行比較；輸出中間結(jié)果＜key，value＞對的形式，該鍵值對的含義是＜中間最小元素行號，中間最小元素記錄行＞。map函數(shù)的偽代碼如下:

圖2 線性時間選擇算法的并行化實(shí)現(xiàn)

3.4 reduce函數(shù)的實(shí)現(xiàn)

reduce函數(shù)的任務(wù)是匯總map函數(shù)的輸出的中間結(jié)果［11］，對map函數(shù)輸出的各自最小元素進(jìn)行比較并找出最小元素，將最小元素加入MIN＿LIST中，此時對k和t的值進(jìn)行比較，如果k＝t那么任務(wù)結(jié)束，要找的元素即為MIN＿LIST中的最后一個元素，否則更新MIN＿LIST后進(jìn)行下一輪的Map ReduceJob任務(wù)。在reduce函數(shù)中，輸入數(shù)據(jù)為＜key，value＞對的形式，該鍵值對的含義是＜中間最小元素行號，中間最小元素記錄行＞；將所有的數(shù)據(jù)送給同一個reduce任務(wù)，對所有的輸入數(shù)據(jù)進(jìn)行比較，計算出最小值，并更新MIN＿LIST描述文件；輸出結(jié)果為＜key，value＞對的形式，該鍵值對的含義是＜最小元素行號，最小元素記錄行＞。reduce函數(shù)的偽代碼如下所示:

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

4.1 實(shí)驗(yàn)環(huán)境搭建

圖3是本實(shí)驗(yàn)中用到的集群結(jié)構(gòu)示意圖，共6臺機(jī)器，每臺機(jī)器的硬件配置如下:CPU型號為Intel Xeon X3330，內(nèi)存為8GB；硬盤2TB SATA；板載Intel雙千兆網(wǎng)絡(luò)控制器。其中1臺機(jī)器作為Job Tracker和Name Node主服務(wù)節(jié)點(diǎn)，其他機(jī)器均作為Task Tracker和Data Node子服務(wù)節(jié)點(diǎn)。根據(jù)Hadoop項目的官方網(wǎng)站相關(guān)的方法配置Hadoop1.0.4版本的集群。

4.2 單機(jī)處理結(jié)果分析實(shí)驗(yàn)

實(shí)驗(yàn)主要完成的內(nèi)容為，比較線性時間選擇算法的串行實(shí)現(xiàn)與Hadoop集群中的一個節(jié)點(diǎn)在相同的硬件環(huán)境配置下［12］，處理相同規(guī)模數(shù)據(jù)，從數(shù)據(jù)的讀入一直到最終的數(shù)據(jù)的導(dǎo)出所要完成的時間。在實(shí)驗(yàn)的過程當(dāng)中，線性時間選擇算法串行采用Java編程語言來實(shí)現(xiàn)，實(shí)驗(yàn)環(huán)境Windows XP，jdk1.7.0＿15。在兩種實(shí)現(xiàn)方法的對比實(shí)驗(yàn)中，處理的數(shù)據(jù)文件大小為5811KB，k的取值從小到大依次遞增，實(shí)驗(yàn)情況如圖4所示，其中LTS表示傳統(tǒng)的線性時間選擇算法實(shí)驗(yàn)結(jié)果，PLTS表示改進(jìn)后的算法實(shí)驗(yàn)結(jié)果。

圖3 集群配置結(jié)構(gòu)

圖4 單機(jī)對比實(shí)驗(yàn)結(jié)果

通過上述的實(shí)驗(yàn)表明:對相同的數(shù)據(jù)文件進(jìn)行操作時，兩種不同的實(shí)現(xiàn)方法在隨著k取值的增長，產(chǎn)生了明顯的對比，在k值較小時單機(jī)串行顯示出了較好的優(yōu)勢，而Hadoop集群上的單節(jié)點(diǎn)的線性時間選擇算法的并行實(shí)現(xiàn)顯得比較笨拙，因?yàn)镠adoop集群中的作業(yè)啟動和任務(wù)分配以及服務(wù)節(jié)點(diǎn)與數(shù)據(jù)節(jié)點(diǎn)之間的交互，都需要耗費(fèi)一定的時間和占用的資源，而單機(jī)上的串行算法所消耗的資源要遠(yuǎn)遠(yuǎn)小于Hadoop集群所消耗的資源，從而在數(shù)據(jù)規(guī)模小的情況下節(jié)約了時間；在k值較大時，Hadoop集群上的單節(jié)點(diǎn)的線性時間選擇算法的并行實(shí)現(xiàn)顯示出它的優(yōu)勢，因?yàn)镠adoop集群具有處理較大規(guī)模數(shù)據(jù)的能力，而傳統(tǒng)的串行實(shí)現(xiàn)因?yàn)檫\(yùn)算的次數(shù)不斷的增加而變得緩慢。

4.3 集群處理結(jié)果及分析

本實(shí)驗(yàn)的目的是在Hadoop集群中進(jìn)行線性時間選擇算法的并行化實(shí)現(xiàn)，在大規(guī)模數(shù)據(jù)處理［13］的情況下，對于k值較大時，Hadoop集群的單節(jié)點(diǎn)實(shí)驗(yàn)已經(jīng)顯示出了它獨(dú)到的優(yōu)勢，那么在Hadoop集群多節(jié)點(diǎn)共同完成線性時間選擇算法的并行化實(shí)現(xiàn)必然會進(jìn)一步提高效率因此，可以進(jìn)行針對更大規(guī)模數(shù)據(jù)的線性時間選擇算法的并行化實(shí)驗(yàn)。本實(shí)驗(yàn)選擇3組數(shù)據(jù)集，如表1所示，每個元素是浮點(diǎn)型數(shù)據(jù)，為了顯示Hadoop集群對大規(guī)模數(shù)據(jù)的處理能力，我們對k采取較大的值，k取值為4000進(jìn)行實(shí)驗(yàn)。

表1 數(shù)據(jù)集情況

實(shí)驗(yàn)中分別選擇1，2，3，4，5個任務(wù)節(jié)點(diǎn)（tasktracker）來參與計算，主要考察的目標(biāo)是在相同規(guī)模數(shù)據(jù)，處理相同問題的情況下，逐漸增加節(jié)點(diǎn)時所需要完成任務(wù)花費(fèi)的時間，實(shí)驗(yàn)結(jié)果如圖5所示。

圖5 線性時間選擇算法Hadoop集群并行化處理結(jié)果

通過實(shí)驗(yàn)可以得出，在相同規(guī)模數(shù)據(jù)，處理相同問題的情況下，隨著節(jié)點(diǎn)數(shù)的增加，完成任務(wù)所需要花費(fèi)的時間就越少，因此增加節(jié)點(diǎn)會顯著提高完成任務(wù)的效率。這足以說明Map Reduce在處理線性時間選擇算法的并行化實(shí)現(xiàn)上有著顯著的優(yōu)勢和可行性。

5 結(jié)束語

本文針對線性時間選擇算法無法在有效的時間內(nèi)處理大數(shù)據(jù)的問題，提出了利用Map Reduce模型并行實(shí)現(xiàn)線性時間選擇算法，將文件中的數(shù)據(jù)分配給各個任務(wù)節(jié)點(diǎn)，各任務(wù)節(jié)點(diǎn)完成局部最優(yōu)解的計算，然后系統(tǒng)將各個局部最優(yōu)解匯總并計算出全局最優(yōu)解。實(shí)驗(yàn)結(jié)果表明，MapReduce模型下的并行線性時間選擇算法，在面對大規(guī)模數(shù)據(jù)時能夠體現(xiàn)出其高效率的優(yōu)勢。

［1］Zhi Yitan，Yong He.Liner time algorithms for parallel machine scheduling［J］.Acta Mathematica Sinica，English Series，2007，23（1）:137－146.

［2］TAN Xiongpai，WANG Huiju，WANG Shan，et al.Big data analysis－competition and symbiosis of RDBMS and MapReduce［J］.Journal of Software，2012，23（1）:32－45（in Chinese）.［覃雄派，王會舉，王珊，等.大數(shù)據(jù)分析——RDBMS與MapReduce的競爭與共生［J］.軟件學(xué)報，2012，23（1）:32－45.］

［3］ZHAO Yanrong，WANG Weiping，MENG Dan，et al.Efficient join query processing algorithm CHMJ based on hadoop［J］.Journal of Software，2012，23（8）:2032－2041（in Chinese）.［趙彥榮，王偉平，孟丹，等.基于Hadoop的高效連接查詢處理算法CHMJ［J］.軟件學(xué)報，2012，23（8）:2032－2041.］

［4］ZHANG Boliang，ZHOU Shuigeng，GUAN Jihong.Skyline computation under MapReduce framework［J］.Journal of Frontiers of Computer Science and Technology，2011，5（5）:385－397（in Chinese）.［張波良，周水庚，關(guān)佶紅.MapReduce框架下的Skyline計算［J］.計算機(jī)科學(xué)與探索，2011，5（5）:385－397.］

［5］Polo J，Carrera D.Performance－driven task co－scheduling for Map Reduce environments［C］//IEEE Network Operations and Management Symposium Conference，2010:373－380.

［6］LI Jianjiang，CUI Jian，WANG Dan，et al.Survey of Map Reduce parallel programming model［J］.Acta Electronica Sinica，2011，39（11）:2635－2642（in Chinese）.［李建江，崔健，王聃，等.Map Reduce并行編程模型研究綜述［J］.電子學(xué)報，2011，39（11）:2635－2642.］

［7］DING Linlin，XIN Junchang，WANG Guoren，et al.Efficient Skyline query processing of massive data based on map－reduce［J］.Journal of Computers，2011，34（10）:1785－1796（in Chinese）.［丁琳琳，信俊昌，王國仁，等.基于Map－Reduce的海量數(shù)據(jù)高效Skyline查詢處理［J］.計算機(jī)學(xué)報，2011，34（10）:1785－1796.］

［8］Shafer J，Rixner S，Cox A L.The hadoop distributed filesystem:Balancing portability and performance［C］//Proceedings of the IEEE International Symposium on Performance Analysis of Systems ＆Software.Washington，DC，USA:IEEE Computer Society，2010:122－133.

［9］PAN Wei，LI Zhanhuai，WU Sai，et al.Evaluatig Large graph processing in Map Reduce based on message passing［J］.Journal of Computers，2011，34（10）:1768－1784（in Chinese）.［潘巍，李戰(zhàn)懷，伍賽，等.基于消息傳遞機(jī)制的MapReduce圖算法研究［J］.計算機(jī)學(xué)報，2011，34（10）:1768－1784.］

［10］Huang D，Shi X，Ibrahim S，et al.MR－Scope:A real－time tracing tool for Map Reduce［C］.The Map Reduce of HPDC，2010:849－855.

［11］Steven J Plimpton，Karen D Devine.MapReduce in MPI for Large－scale graph algorithms［J］.Parallel Computing，2011，37（9）:610－632.

［12］LUAN Yajian，HUANG Chongmin，GONG Gaosheng，et al.Research on performance optimization of hadoop platform［J］.Computer Engineering，2010，36（14）:262－264（in Chinese）.［欒亞建，黃翀民，龔高晟，等.Hadoop平臺的性能優(yōu)化研究［J］.計算機(jī)工程，2010，36（14）:262－264.］

［13］Fabrizio Marozzo，Domenico Talia，Paolo Trunfio.P2P－Map Reduce:Parallel data processing in dynamic cloud environments［J］.Journal of Computer and System Sciences，2011，78（5）:1382－1402.