MapReduce迭代式計(jì)算發(fā)展與應(yīng)用

馬在營　劉建新　徐彬

摘要摘要：MapReduce的優(yōu)勢集中體現(xiàn)在并行計(jì)算上，而在迭代計(jì)算上則存在諸多不足。科研人員不斷對MapReduce并行計(jì)算模型進(jìn)行迭代計(jì)算優(yōu)化，使MapReduce可以支持顯示迭代式計(jì)算。介紹了傳統(tǒng)MapReduce框架與迭代式MapReduce框架，通過K-means算法測試了iMapReduce、Hadoop MapReduce的迭代性能，給出了實(shí)驗(yàn)結(jié)果及分析。

關(guān)鍵詞關(guān)鍵詞：迭代式計(jì)算；MapReduce；Haloop；Hadoop；iMapReduce

DOIDOI：10.11907/rjdk.162775

中圖分類號：TP301

文獻(xiàn)標(biāo)識碼：A文章編號文章編號：16727800（2017）005020703

0引言

在全球信息產(chǎn)業(yè)高速發(fā)展融合的背景下，網(wǎng)絡(luò)數(shù)據(jù)資源規(guī)模急劇膨脹，尤其是高能物理、互聯(lián)網(wǎng)應(yīng)用、基因工程、電子商務(wù)以及計(jì)算機(jī)仿真等領(lǐng)域的數(shù)據(jù)量攀升速度驚人，現(xiàn)有數(shù)據(jù)分析工具很難滿足日益增長的海量密集型數(shù)據(jù)的信息處理需求。基于此，Google實(shí)驗(yàn)室專門設(shè)計(jì)了MapReduce編程模型，該模型在數(shù)據(jù)信息的批量處理上優(yōu)勢明顯，但在迭代處理上問題也相當(dāng)突出。逐次逼近是迭代計(jì)算的基本思想，迭代計(jì)算過程為先取近似值，然后采用遞推公式對近似值反復(fù)校正，直至精度達(dá)到要求為止。當(dāng)數(shù)據(jù)量較小時(shí)，可以在單機(jī)上進(jìn)行迭代計(jì)算，但當(dāng)數(shù)據(jù)量非常大時(shí)，迭代處理則極其耗時(shí)。在數(shù)據(jù)挖掘、信息檢索、機(jī)器學(xué)習(xí)等領(lǐng)域，有很多算法需要迭代，傳統(tǒng)的迭代計(jì)算不能有效應(yīng)對當(dāng)前的大數(shù)據(jù)處理需求。經(jīng)過幾年時(shí)間，科研工作者對MapReduce進(jìn)行迭代計(jì)算改進(jìn)，產(chǎn)生了一些迭代式計(jì)算框架，包括比較知名的Twister、Haloop等。

1傳統(tǒng)MapReduce模型框架

Google于2004年在論文《分布式計(jì)算：基于大型集群的數(shù)據(jù)簡化處理》中首次提出MapReduce框架，指出MapReduce框架在處理密集型應(yīng)用數(shù)據(jù)過程中將處理程序簡化抽象為Map與Reduce兩個(gè)階段，用戶在進(jìn)行分布式程序設(shè)計(jì)過程中，只需調(diào)用reduce（）和map（）兩個(gè)函數(shù)即可，無需過多考慮任務(wù)調(diào)度、設(shè)備通信、數(shù)據(jù)分片以及容錯(cuò)等細(xì)節(jié)問題。在MapReduce框架內(nèi)，這些問題都能得到很好的處理。作為MapReduce框架的主要思想，Map與Reduce來自函數(shù)編程語言，其原理如圖1所示。Map將數(shù)據(jù)打散，Reduce則負(fù)責(zé)聚集這些數(shù)據(jù)。在Map環(huán)節(jié)，maptask每讀取一個(gè)block，即采用map（）函數(shù)對數(shù)據(jù)進(jìn)行處理，同時(shí)將處理結(jié)果寫入本地磁盤；在Reduce環(huán)節(jié)，每個(gè)reduce task在對Map Task節(jié)點(diǎn)數(shù)據(jù)進(jìn)行遠(yuǎn)程讀取過程中，都會采用reduce（）函數(shù)處理數(shù)據(jù)，同時(shí)將處理完成的數(shù)據(jù)寫入分布式文件系統(tǒng)內(nèi)。在MapReduce框架內(nèi)，用戶只要通過map和reduce端口，即可快速計(jì)算TB級數(shù)據(jù)，如數(shù)據(jù)挖掘和日志分析等常見應(yīng)用。此外，MapReduce框架還適用于圓周率等科學(xué)數(shù)據(jù)的計(jì)算。

通過上述分析可知，在傳統(tǒng)MapReduce框架內(nèi)，無論是map環(huán)節(jié)，還是reduce環(huán)節(jié)，數(shù)據(jù)處理結(jié)果均需寫入磁盤內(nèi)，雖然這一過程會降低系統(tǒng)性能，但是能夠提高系統(tǒng)可靠性。也正因如此，傳統(tǒng)MapReduce在迭代計(jì)算處理上問題突出，若用戶強(qiáng)行在傳統(tǒng)MapReduce上進(jìn)行迭代計(jì)算，系統(tǒng)性能則會變得非常差。

2迭代式MapReduce框架

2.1Twister

在Twister內(nèi)，大文件不會被自動切割為單個(gè)block，所以用戶必須提前將文件分割為小文件才能進(jìn)行task處理。在map環(huán)節(jié)，調(diào)用map（）函數(shù)處理后，數(shù)據(jù)結(jié)果存儲于分布式內(nèi)存之中，然后利用broker network將數(shù)據(jù)推送至reduce task（若Twister內(nèi)存較大，則所有中間數(shù)據(jù)都能存儲其中）；在reduce環(huán)節(jié)，通過combine對reducetask產(chǎn)生的結(jié)果進(jìn)行歸并，此時(shí)用戶可依據(jù)條件作出是否結(jié)束迭代計(jì)算的決定。合并后的數(shù)據(jù)被分解傳送至map task，進(jìn)行新一輪迭代計(jì)算。為有效提高系統(tǒng)容錯(cuò)性，Twister會定時(shí)將reducetask和maptask產(chǎn)生的結(jié)果錄入磁盤內(nèi)，避免task失敗后數(shù)據(jù)丟失。即使task失敗，依然可以從磁盤內(nèi)調(diào)取數(shù)據(jù)重新進(jìn)行迭代處理。Twister架構(gòu)如圖2所示。

為避免用戶在迭代計(jì)算中對task的重建，Twister建立了一個(gè)taskpool，當(dāng)用戶需要使用task時(shí)，可直接從pool中讀取。在Twister內(nèi)，所有數(shù)據(jù)與消息均由broker network傳遞，brokernetwork是獨(dú)立模塊，現(xiàn)階段僅支持ActiveMQ和NaradaBroking。目前Twister尚屬于研究性項(xiàng)目，其設(shè)計(jì)策略決定了Twister還不能在實(shí)際中得到廣泛應(yīng)用。例如Twister將數(shù)據(jù)存儲于分布式內(nèi)存中，而缺乏分布式文件系統(tǒng)，僅提供tool對文件進(jìn)行訪問與存儲，并存在迭代計(jì)算模型不夠抽象、支持應(yīng)用類型偏少等問題。

2.2Haloop

Haloop是Haloop Mapreduce 的修改版，Haloop不僅擴(kuò)展了分布迭代編程，并通過使用任務(wù)調(diào)度器loopaware添加各種緩存機(jī)制，大大提高了效率。其架構(gòu)如圖3所示。

圖3有3個(gè)工作在運(yùn)行，工作1、工作2、工作3。每個(gè)工作都有3個(gè)任務(wù)同時(shí)從slave節(jié)點(diǎn)上運(yùn)行。為了適應(yīng)迭代計(jì)算，Haloop以Hadoop為基礎(chǔ)作了一些調(diào)整：①Haloop提供了一個(gè)新的應(yīng)用程序用戶編程接口，簡化了迭代表達(dá)式的表達(dá)；②Haloop的master節(jié)點(diǎn)包含一個(gè)新的loop控制模塊，可以指定迭代停止條件；③Haloop為迭代計(jì)算使用一個(gè)新的任務(wù)調(diào)度器，可以將數(shù)據(jù)本地化；④Haloop的緩存和索引應(yīng)用程序在slave節(jié)點(diǎn)上。

（1）在Haloop內(nèi)迭代式任務(wù)全部被抽象為：

Ro代表初始輸入，Ri代表第i次迭代的結(jié)果，L是迭代計(jì)算中保持不變的數(shù)據(jù)。Haloop主要編程接口為：

SetFixedPointThreshold：設(shè)置好迭代計(jì)算的結(jié)束條件，即距離差的閾值。

Set the number of iterations：設(shè)置迭代次數(shù)。

Setiterationinput：設(shè)置迭代變化輸入數(shù)據(jù)。

AddinvariantTable：設(shè)置不變的輸入數(shù)據(jù)。

（2） Loop-aware 任務(wù)調(diào)度。 Haloop 在初次迭代計(jì)算中會將不變的輸入數(shù)據(jù)存儲于計(jì)算節(jié)點(diǎn)中，以便后續(xù)的task調(diào)度，且數(shù)據(jù)應(yīng)當(dāng)盡量存儲于locality等固定節(jié)點(diǎn)中，使每次迭代計(jì)算時(shí)無需重新傳輸數(shù)據(jù)。

（3）Index與Cache。無論是map task的輸入還是輸出，reducetask的輸出都會通過緩存與建索引來加快迭代計(jì)算速度。其中，緩存主要指迭代計(jì)算結(jié)果被寫入磁盤以供循環(huán)迭代使用。

總體來看，與Twister相比，Haloop更為抽象，能夠支持多種計(jì)算。此外，Haloop是在Hadoop基礎(chǔ)上優(yōu)化改進(jìn)而成，因此具備Hadoop的優(yōu)點(diǎn)。

2.3iMapReduce

張巖峰在論文《iMapReduce：分布式計(jì)算框架迭代計(jì)算》中提出iMapReduce概念，iMapReduce是以盡量減少系統(tǒng)開銷為目的而構(gòu)建的高效迭代計(jì)算框架。MapReduce需要一系列的迭代作業(yè)來處理迭代計(jì)算，圖4左圖是MapReduce迭代處理過程。在iMapReduce中，迭代任務(wù)只發(fā)生在初始階段和結(jié)束階段，用戶僅需建立一個(gè)作業(yè)任務(wù)即能完成迭代計(jì)算，避免了對系統(tǒng)的反復(fù)開銷。通過對本地靜態(tài)數(shù)據(jù)的維護(hù)，減少因反復(fù)加載數(shù)據(jù)而產(chǎn)生的系統(tǒng)開銷，同時(shí)允許在單次迭代計(jì)算中異步執(zhí)行map任務(wù)。iMapReduce極大地提高了迭代算法性能。圖4展示了iMapReduce的迭代計(jì)算過程。iMapReduce為用戶提供編程的API接口。iMapReduce兼容Hadoop MapReduce任務(wù)，用戶可根據(jù)實(shí)際決定是否啟用iMapReduce功能。iMapReduce將靜態(tài)數(shù)據(jù)和迭代數(shù)據(jù)區(qū)分開。圖5是iMapReduce迭代計(jì)算節(jié)點(diǎn)數(shù)據(jù)流，在初始環(huán)節(jié)中，靜態(tài)數(shù)據(jù)與迭代數(shù)據(jù)被加載至本地文件系統(tǒng)，其中靜態(tài)數(shù)據(jù)始終存儲于本地文件系統(tǒng)內(nèi)，而迭代數(shù)據(jù)在調(diào)用map（）函數(shù)處理前，需要同靜態(tài)數(shù)據(jù)聯(lián)合執(zhí)行操作，即將靜態(tài)數(shù)據(jù)與迭代數(shù)據(jù)連接起來，作為map（）函數(shù)輸入。

通過這些迭代優(yōu)化，使iMapReduce更加適合大數(shù)據(jù)迭代處理，并且提供了編程接口API，兼容HadoopMapreduce任務(wù)，用戶可選擇是否啟用iMapReduce的迭代處理功能。

3MapReduce迭代計(jì)算迭代性能測試

3.1實(shí)驗(yàn)設(shè)計(jì)

集群環(huán)境由5臺計(jì)算機(jī)組成（1個(gè)Master節(jié)點(diǎn)，4個(gè)Slave節(jié)點(diǎn)），系統(tǒng)為64位Windows7，處理器為i7-4790 3.60GHz，內(nèi)存8GB，網(wǎng)速10Mb/s。

K-means 算法是聚類分析中使用最廣泛的算法之一，可用于測試不同平臺的迭代性能。已知觀測集（x1，x2，x3，...，xn），其中每個(gè)觀測都是一個(gè)d維實(shí)向量，k-平均聚類要將這n個(gè)觀測劃分到k個(gè)集合中（k≤n），使組內(nèi)平方和（WCSS，Within-Cluster Sum of Squares）最小。換言之，它的目標(biāo)是找到使式（1）滿足的聚類Si，其中μi是Si中所有點(diǎn)的均值。簡單描述為：不斷迭代計(jì)算各個(gè)數(shù)據(jù)簇的中心點(diǎn)，直到該中心點(diǎn)趨于穩(wěn)定。

argminS=∑i=k∑x∈Six-μi2（1）

K-means步驟如下：①創(chuàng)建k個(gè)數(shù)據(jù)簇的中心點(diǎn)；②計(jì)算所有數(shù)據(jù)點(diǎn)到k個(gè)中心點(diǎn)的距離，將其劃歸到距離自己最近的中心點(diǎn)；③根據(jù)上次聚類結(jié)果，計(jì)算各個(gè)數(shù)據(jù)簇的算數(shù)平均值作為新的數(shù)據(jù)簇中心點(diǎn)；④將所有數(shù)據(jù)在新中心點(diǎn)上重新聚類；⑤重復(fù)第4步，直到中心點(diǎn)趨于穩(wěn)定。

通過測試K-means算法在Hadoop Mapreduce、iMapReduce的運(yùn)行情況，統(tǒng)計(jì)出不同平臺運(yùn)行K-means的時(shí)間。

3.2實(shí)驗(yàn)結(jié)果

從表1中可以看出，iMapReduce運(yùn)行時(shí)間遠(yuǎn)遠(yuǎn)短于Hadoop MapReduce，大約是Hadoop MapReduce運(yùn)行時(shí)間的25%～39%，這是由于Hadoop不支持迭代計(jì)算，花了大量時(shí)間在數(shù)據(jù)通信上，從而大大降低了執(zhí)行速度。而iMapReduce通過避免傳輸靜態(tài)數(shù)據(jù)而減少了通信開銷，并在一次迭代內(nèi)允許異步執(zhí)行，因此對于迭代應(yīng)用有良好的性能表現(xiàn)。

4結(jié)語

迭代式計(jì)算是大數(shù)據(jù)領(lǐng)域中一種重要的計(jì)算方法，Twister和Haloop的模型抽象度不夠高，支持的計(jì)算有限，iMapReduce有效提升了大數(shù)據(jù)迭代計(jì)算的性能。目前MapReduce迭代式計(jì)算還處于發(fā)展階段，在以后的研究中，如何優(yōu)化提高分布式迭代計(jì)算的作業(yè)調(diào)度速度是亟待解決的問題。

參考文獻(xiàn)參考文獻(xiàn)：

[1]ZHANG Y， GAO Q， GAO L， et al. iMapReduce： a distributed computing framework for iterative computation[J]. Journal of Grid Computing， 2012， 10（1）：11121121.

[2]EKANAYAKE J， LI H， ZHANG B， et al. Twister： a runtime for iterative MapReduce[C].ACM International Symposium on High Performance Distributed Computing，2010：810818

[3]Bu Y， Howe B， Balazinska M， et al. Haloop： efficient iterative data processing on large clusters[J]. Proceedings of the Vldb Endowment， 2010， 3（12）：285296.

[4]張巖峰.云環(huán)境下大數(shù)據(jù)迭代計(jì)算研究[J].沈陽：東北大學(xué)，2012.

[5]易修文， 李天瑞， 張鈞波，等. 不同MapReduce運(yùn)行系統(tǒng)的性能測試與分析[J].計(jì)算機(jī)科學(xué)， 2015，42（5）：2427.

[6]董的博客. 傳統(tǒng)MapReduce框架介紹[EB/OL]. http：//dongxicheng.org/mapreduce/traditionalmapreduceframework/.

[7]Kmeans算法的Hadoop實(shí)現(xiàn)[EB/OL].http：//blog.csdn.net/chaaaa_wangyc/article/details/53426612？locationNum=9&fps=1.

責(zé)任編輯（責(zé)任編輯：黃健）