葛君偉，蔣 仙，方義秋

1.重慶郵電大學(xué) 圖書(shū)館，重慶 4000652.重慶郵電大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，重慶 400065

消息代理機(jī)制下的MapReduce數(shù)據(jù)流優(yōu)化

葛君偉1，蔣 仙2，方義秋2

1.重慶郵電大學(xué) 圖書(shū)館，重慶 400065
2.重慶郵電大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，重慶 400065

1 概述

隨著當(dāng)前信息時(shí)代的迅速發(fā)展，數(shù)據(jù)量日益增大，數(shù)據(jù)處理流程呈現(xiàn)出海量和并行的特點(diǎn)。傳統(tǒng)的并行技術(shù)如網(wǎng)格計(jì)算，由于編程難度大并且處理流程復(fù)雜，無(wú)法滿足當(dāng)前數(shù)據(jù)密集型計(jì)算DIC（Data Intensive Computing）應(yīng)用的需求。目前各研究機(jī)構(gòu)（如：Google、Amazon）在解決專門問(wèn)題領(lǐng)積累了重要的知識(shí)基礎(chǔ)，隨之新型的云計(jì)算環(huán)境下的并行處理技術(shù)也應(yīng)運(yùn)而生[1]。

其中，較為流行的是由Google提出的MapReduce并行計(jì)算編程模型，極大地簡(jiǎn)化了在集群上的海量數(shù)據(jù)處理過(guò)程。主流的Apache Hadoop開(kāi)源開(kāi)放平臺(tái)，為MapReduce應(yīng)用提供了較好的實(shí)現(xiàn)平臺(tái)，從而促使當(dāng)前大量商用的海量數(shù)據(jù)分析采用MapReduce并行計(jì)算模型。

但另一方面，MapReduce編程模型的簡(jiǎn)單化也導(dǎo)致了一些問(wèn)題，尤其是面向數(shù)據(jù)密集型計(jì)算。由于此類應(yīng)用在很大程度上依賴于集群節(jié)點(diǎn)間的數(shù)據(jù)傳輸，主節(jié)點(diǎn)對(duì)節(jié)點(diǎn)間的消息處理負(fù)載過(guò)重，最終造成運(yùn)行時(shí)間長(zhǎng)、計(jì)算結(jié)果值不理想等問(wèn)題。

而消息代理機(jī)制（Message Broker Mechanism）是一種在數(shù)據(jù)源與目的地之間移動(dòng)數(shù)據(jù)使信息處理流暢的消息中間件技術(shù)。為了滿足云計(jì)算環(huán)境下對(duì)DIC應(yīng)用的可靠性、可擴(kuò)展性等方面的要求，在MapReduce并行計(jì)算框架中采用消息代理機(jī)制，兼具兩者優(yōu)點(diǎn)，并行計(jì)算簡(jiǎn)單化，消息處理獨(dú)立化。

本文將消息代理機(jī)制應(yīng)用于MapReduce框架中，采用消息代理和消息隊(duì)列技術(shù)，優(yōu)化數(shù)據(jù)流，解決面向數(shù)據(jù)密集型計(jì)算的MapReduce應(yīng)用的負(fù)載均衡問(wèn)題。

2 MapReduce

MapReduce框架把一些DIC應(yīng)用的數(shù)據(jù)處理過(guò)程簡(jiǎn)化抽象成map和reduce兩個(gè)階段，用戶在設(shè)計(jì)分布式程序時(shí)，只要實(shí)現(xiàn)Mapper和Reducer兩個(gè)函數(shù)，將數(shù)據(jù)分片，任務(wù)調(diào)度，集群容錯(cuò)，集群間通信等問(wèn)題，交由底層的MapReduce框架處理。

2.1 MapReduce體系架構(gòu)

Google的研究人員借鑒函數(shù)式編程的設(shè)計(jì)思想，通過(guò)總結(jié)大量的大規(guī)模分布式處理程序共同特征，提出了MapReduce并行編程框架，架構(gòu)圖如圖1。

圖1 MapReduce架構(gòu)示意圖

從圖1可以看到，MapReduce體系結(jié)構(gòu)中主要包括了一個(gè)Master節(jié)點(diǎn)和若干個(gè)Worker節(jié)點(diǎn)。Master Node有且只有一個(gè)，主要的工作是統(tǒng)一調(diào)度并且維護(hù)每個(gè)Worker Node的工作狀態(tài)，具有整個(gè)架構(gòu)中的控制中心；Worker Node有多個(gè)，其中 Mapper-Worker節(jié)點(diǎn)，完成map任務(wù)產(chǎn)生中間結(jié)果，Reducer-Worker節(jié)點(diǎn)完成reduce任務(wù)輸出最終結(jié)果。具體數(shù)據(jù)處理過(guò)程見(jiàn)文獻(xiàn)[2]。

從整個(gè)數(shù)據(jù)流程可以看出，MapReduce模型較好地實(shí)現(xiàn)了并行計(jì)算的數(shù)據(jù)處理，而且數(shù)據(jù)的分發(fā)、計(jì)算及傳輸對(duì)用戶是透明的。

2.2 MapReduce DIC應(yīng)用

基于MapReduce實(shí)現(xiàn)的DIC應(yīng)用相比于傳統(tǒng)計(jì)算方法，具有高效及可靠的海量數(shù)據(jù)處理優(yōu)勢(shì)。

DIC應(yīng)用需要處理的數(shù)據(jù)具有以下幾個(gè)特點(diǎn)：

（1）數(shù)據(jù)規(guī)模是海量的，達(dá)到了TB甚至PB量級(jí)。

（2）數(shù)據(jù)形式是多樣的，往往不是真正的關(guān)系型數(shù)據(jù)（或稱為結(jié)構(gòu)化數(shù)據(jù)），而可能是文本、網(wǎng)頁(yè)、XML文檔等無(wú)結(jié)構(gòu)或者半結(jié)構(gòu)化數(shù)據(jù)。

（3）數(shù)據(jù)分布是非集中式的，存儲(chǔ)于不同地理位置的多機(jī)架上。

MapReduce應(yīng)用能自動(dòng)完成數(shù)據(jù)密集型任務(wù)的分發(fā)和并行處理，如Facebook的服務(wù)事務(wù)日志分析、用戶操作分析、用戶點(diǎn)擊量分析等計(jì)算，Yahoo!部署Hadoop、PIG促進(jìn)開(kāi)源項(xiàng)日研究[1]。

3 分析與設(shè)計(jì)

本文的核心是在MapReduce模型框架中引入消息代理實(shí)體，管理控制Reduce-Worker傳來(lái)的接收消息；標(biāo)記Mapper-Worker產(chǎn)生的中間結(jié)果，利用基于標(biāo)簽的消息隊(duì)列實(shí)現(xiàn)異步傳輸模式，從而實(shí)現(xiàn)數(shù)據(jù)流的優(yōu)化。

3.1 消息代理機(jī)制

3.2 基于消息代理機(jī)制的MapReduce模型框架

傳統(tǒng)的MapReduce框架中Master Node依賴來(lái)自每個(gè)Worker Node的定期心跳（heartbeat）消息。每條消息都包含一個(gè)數(shù)據(jù)塊報(bào)告，Master Node可以根據(jù)這個(gè)報(bào)告驗(yàn)證塊映射和其他文件系統(tǒng)元數(shù)據(jù)。如果Worker Node不能發(fā)送心跳消息，Master Node將采取修復(fù)措施，重新復(fù)制在該節(jié)點(diǎn)上丟失的塊。然而，通常數(shù)據(jù)密集型應(yīng)用會(huì)隨需要處理信息數(shù)據(jù)量的增大，Master Node對(duì)Worker Node的管控的工作負(fù)荷也隨之而加重，節(jié)點(diǎn)間的消息處理負(fù)載過(guò)重，導(dǎo)致中間結(jié)果數(shù)據(jù)傳輸缺乏可靠性、負(fù)載不均，最終使得運(yùn)行效果不理想。為此，將消息處理模塊以獨(dú)立實(shí)體形式存在于集群中，使得消息處理獨(dú)立化，實(shí)現(xiàn)中間結(jié)果數(shù)據(jù)可靠傳遞。

圖2 基于消息代理機(jī)制的MapReduce模型框架示意圖

基于消息代理機(jī)制的MapReduce模型中由Mapper-Worker通過(guò)map函數(shù)產(chǎn)生的中間結(jié)果值都被唯一標(biāo)識(shí)。核心部分是由消息代理（Message Broker，MB）和基于標(biāo)識(shí)的傳輸消息隊(duì)列兩大部分組成。MB以獨(dú)立實(shí)體存在于分布式集群中，管控節(jié)點(diǎn)消息通信。而消息隊(duì)列則駐留在Worker Node的內(nèi)存中便于其各個(gè)節(jié)點(diǎn)的數(shù)據(jù)塊傳遞。同時(shí)，各個(gè)接口實(shí)現(xiàn)消息數(shù)據(jù)的傳遞與處理。

消息代理的主要功能是對(duì)消息進(jìn)行管理與控制，采用一對(duì)一的消息傳送方式。其工作原理是：首先，當(dāng)一個(gè)實(shí)例作業(yè)（job）開(kāi)始執(zhí)行時(shí)就會(huì)創(chuàng)建一個(gè)守護(hù)進(jìn)程-MB進(jìn)程和其主線程；然后，一旦任務(wù)分配后，各個(gè)任務(wù)（task）創(chuàng)建各自的獨(dú)立線程，這些線程將各個(gè)節(jié)點(diǎn)發(fā)來(lái)的消息交由主線程處理。當(dāng)Reduce任務(wù)執(zhí)行完畢對(duì)應(yīng)的獨(dú)立線程也將轉(zhuǎn)為結(jié)束狀態(tài)。MB會(huì)把消息劃分為三類，臨時(shí)消息、持久消息和一般消息。一旦臨時(shí)消息被處理完將立即刪除，而持久消息則存儲(chǔ)于MB的指定文件中，一般消息將駐留在內(nèi)存，直到對(duì)應(yīng)的MB線程結(jié)束。消息代理的優(yōu)點(diǎn)表現(xiàn)在一方面，消息代理以獨(dú)立于的實(shí)體存在，便于用戶對(duì)作業(yè)運(yùn)行狀態(tài)的動(dòng)態(tài)跟蹤與實(shí)時(shí)監(jiān)控；另一方面，消息代理記錄存儲(chǔ)任務(wù)執(zhí)行信息，其數(shù)據(jù)可用于應(yīng)用性能分析與進(jìn)一步改進(jìn)。

MB將處理完成后的消息都捎帶一個(gè)標(biāo)識(shí)符，此標(biāo)識(shí)符與各自中間結(jié)果值的標(biāo)識(shí)符一致。這些消息通過(guò)MB與主節(jié)點(diǎn)的通信接口，傳遞到主節(jié)點(diǎn)的預(yù)留緩存區(qū)，對(duì)應(yīng)的標(biāo)識(shí)符也將送入到消息隊(duì)列中。Master Node的JobTracker進(jìn)程將從隊(duì)列中遵循先進(jìn)先出讀取各個(gè)消息，并依據(jù)消息內(nèi)容分配調(diào)度TaskTracker進(jìn)程需要的資源。讀取后的消息在Master Node調(diào)度完各個(gè)reduce任務(wù)之后再次返回給MB記錄并銷毀。

基于消息代理機(jī)制的MapReduce模型下，相比于傳統(tǒng)模型在數(shù)據(jù)流上有所改變，從圖3可對(duì)比得出。

圖3 MapReduce數(shù)據(jù)流對(duì)比示意圖

由圖3可以看出，傳統(tǒng)的MapReduce數(shù)據(jù)流（示意圖1）可簡(jiǎn)單概述為：首先，將存儲(chǔ)于分布式存儲(chǔ)系統(tǒng)中的每個(gè)輸入文件從各個(gè)map工作節(jié)點(diǎn)中獲?。≒ull）而來(lái)，進(jìn)行map函數(shù)計(jì)算。然后，中間結(jié)果值以推（Push）方式存儲(chǔ)于本地系統(tǒng)。同樣reduce工作節(jié)點(diǎn)以遠(yuǎn)程傳輸方式獲?。≒ull）數(shù)據(jù)源，再將計(jì)算完的最終結(jié)果值推（Push）入分布式存儲(chǔ)系統(tǒng)。整個(gè)MapReduce的數(shù)據(jù)流處理對(duì)于每個(gè)工作節(jié)點(diǎn)來(lái)說(shuō)都是同步實(shí)現(xiàn)的。而基于消息代理機(jī)制的框架下（示意圖2），由于消息代理對(duì)消息的獨(dú)立處理，使得數(shù)據(jù)流在中間結(jié)果值到reduce工作節(jié)點(diǎn)變?yōu)榱水惒絺鬏敺绞?，本地存?chǔ)系統(tǒng)可以主動(dòng)將數(shù)據(jù)推送（Push），也可以是reduce工作節(jié)點(diǎn)主動(dòng)獲?。≒ull）數(shù)據(jù)。從數(shù)據(jù)流角度看，大大減緩了數(shù)據(jù)傳輸上的壓力。

4 實(shí)驗(yàn)測(cè)試

本實(shí)驗(yàn)的測(cè)試環(huán)境是由美國(guó)印第安納州大學(xué)India. Futuregrid.Org（https：//portal.futuregrid.org/）提供的Eucalptus云計(jì)算IaaS（Infrastructure as a Service）服務(wù)，其中使用到的硬件配置清單見(jiàn)表1。實(shí)驗(yàn)的運(yùn)行環(huán)境是由一組實(shí)例化虛擬集群組成，通過(guò)配置其鏡像系統(tǒng)（Linux 2.6.x），自主搭建的Hadoop[3]（Apache Hadoop-0.20.2版本）集群（本實(shí)驗(yàn)選的集群大小為10個(gè)虛擬鏡像），NarabaBroker（4.4.2版本）為消息代理實(shí)體。

本實(shí)驗(yàn)通過(guò)Last.fm網(wǎng)絡(luò)音樂(lè)電臺(tái)對(duì)用戶提供的公共API接口，獲取其音樂(lè)家信息元數(shù)據(jù)集，在此數(shù)據(jù)集上計(jì)算各個(gè)音樂(lè)家之間的相似度。此實(shí)例在兩個(gè)不同的計(jì)算環(huán)境上運(yùn)行，Sim-c1-MB（MB運(yùn)行的節(jié)點(diǎn)配置為c1）和Sim-m1-MB（MB運(yùn)行的節(jié)點(diǎn)配置為m1），用Linux系統(tǒng)下命令記錄它們?cè)?min、5 min和15 min時(shí)間間隔的平均負(fù)載變化，如圖4、5。

圖4 Sim-c1-MB的平均負(fù)載

圖5 Sim-m1-MB的平均負(fù)載

隨著計(jì)算的運(yùn)行，消息代理支配節(jié)點(diǎn)間的數(shù)據(jù)傳輸和資源調(diào)度，各時(shí)間點(diǎn)的進(jìn)程狀態(tài)是動(dòng)態(tài)變換的（等待或運(yùn)行），因此集群的平均負(fù)載值也隨之波動(dòng)，直到計(jì)算結(jié)束波動(dòng)曲線也趨近為零。

比較圖4和圖5可以看到，Sim-c1-MB的平均負(fù)載整體上遠(yuǎn)大于Sim-m1-MB，即排隊(duì)等待運(yùn)行的平均進(jìn)程數(shù)遠(yuǎn)大于Sim-m1-MB。圖4中Sim-c1-MB的等待運(yùn)行平均進(jìn)程數(shù)為3.35，最大進(jìn)程數(shù)為5.01，而圖5中Sim-m1-MB分別為0.79和5.01。兩者的運(yùn)行時(shí)間分別為352 s和237 s。

從實(shí)驗(yàn)中可以得出，數(shù)據(jù)密集型計(jì)算的性能依賴于消息代理的配置；消息代理的消息處理能力直接影響節(jié)點(diǎn)間的數(shù)據(jù)傳輸效率。實(shí)驗(yàn)證明，基于消息代理機(jī)制的MapReduce框架能提高數(shù)據(jù)密集型應(yīng)用上的負(fù)載均衡。

5 總結(jié)與展望

綜上所述，面對(duì)處理海量數(shù)據(jù)處理，基于消息代理機(jī)制的MapReduce改進(jìn)模型在運(yùn)行效率上有一定的提高。但由于本文只針對(duì)MapReduce的數(shù)據(jù)密集型應(yīng)用進(jìn)行數(shù)據(jù)流優(yōu)化與實(shí)驗(yàn)分析，具有一定的局限性，并且觀察該模型框架在其他問(wèn)題域的應(yīng)用性能研究有待進(jìn)一步展開(kāi)。

表1 硬件配置清單

[1]孫兆玉，袁志平，黃宇光.面向數(shù)據(jù)密集型計(jì)算Hadoop及其應(yīng)用研究[C]//2008年全國(guó)高性能計(jì)算學(xué)術(shù)年會(huì).無(wú)錫：[s.n.]，2008.

[2]Dean J，Ghemawat S.MapReduce：simplified data processing on large clusters[C]//Proceedings of the 6th Symposium on OperatingSystem DesignandImplementation.New York：ACM Press，2004：137-150.

[3]Apache Hadoop.Hadoop[EB/OL].[2009-03-06].http：//hadoop. apache.org/.

[4]Condie T，Conway N，Alvaro P，et al.MapReduce online，UCB/EECS-2009-136[R].Berkeley：EECS Department，University of California，2009.

[5]戴俊，朱曉民.基于ActiveMQ的異步消息總線的設(shè)計(jì)與實(shí)現(xiàn)[J].計(jì)算機(jī)系統(tǒng)應(yīng)用，2010，19（8）：254-257.

[6]李樂(lè)平，吳泉源.消息代理中間件InforBroker中集群技術(shù)的應(yīng)用[J].微計(jì)算機(jī)信息，2006，22（27）：140-142.

GE Junwei1,JIANG Xian2,FANG Yiqiu2

1.Department of Library,Chongqing University of Posts and Telecommunications,Chongqing 400065,China
2.College of Computer Science and Technology,Chongqing University of Posts and Telecommunications,Chongqing 400065,China

MapReduce programming model is a kind of parallel computing framework which is distributed under the environment of mass data processing system.Currently,the MapReduce applications are widely used for commercial data intensive computing,the data transmission between the nodes on cluster has a large extent dependence.It causes that the load of message handling between the nodes is heavy.This paper puts forward an improved model of MapReduce based on message broker mechanism,to optimize the MapReduce data flow.The experimental data indicates that based on message broker mechanism the MapReduce framework can improve the load balance in data intensive applications.

message broker;MapReduce;data intensive computing;data flow

MapReduce編程模型是廣泛應(yīng)用于云計(jì)算環(huán)境下處理海量數(shù)據(jù)的一種并行計(jì)算框架。然而該框架下的面向數(shù)據(jù)密集型計(jì)算，集群節(jié)點(diǎn)間的數(shù)據(jù)傳輸依賴性較強(qiáng)，造成節(jié)點(diǎn)間的消息處理負(fù)載過(guò)重。提出基于消息代理機(jī)制的MapReduce改進(jìn)模型，優(yōu)化數(shù)據(jù)流。經(jīng)實(shí)驗(yàn)數(shù)據(jù)表明，基于消息代理機(jī)制的MapReduce框架能提高數(shù)據(jù)密集型應(yīng)用上的負(fù)載均衡。

消息代理；MapReduce；數(shù)據(jù)密集型計(jì)算；數(shù)據(jù)流

A

TP391

10.3778/j.issn.1002-8331.1107-0421

GE Junwei,JIANG Xian,FANG Yiqiu.Optimization of MapReduce data flow with message broker mechanism.Computer Engineering and Applications,2013,49（5）：120-122.

葛君偉（1961—），男，博士，教授，主要研究領(lǐng)域?yàn)樵朴?jì)算、軟件工程；蔣仙（1987—），女，碩士研究生，主要研究領(lǐng)域?yàn)椴⑿杏?jì)算；方義秋（1963—），女，副教授，主要研究領(lǐng)域?yàn)橹虚g件技術(shù)。E-mail：gejw@cqupt.edu.cn

2011-07-20

2012-01-13

1002-8331（2013）05-0120-03