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

葛君偉，蔣 仙，方義秋

1.重慶郵電大學 圖書館，重慶 4000652.重慶郵電大學 計算機科學與技術學院，重慶 400065

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

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

1.重慶郵電大學 圖書館，重慶 400065
2.重慶郵電大學 計算機科學與技術學院，重慶 400065

1 概述

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

其中，較為流行的是由Google提出的MapReduce并行計算編程模型，極大地簡化了在集群上的海量數據處理過程。主流的Apache Hadoop開源開放平臺，為MapReduce應用提供了較好的實現平臺，從而促使當前大量商用的海量數據分析采用MapReduce并行計算模型。

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

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

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

2 MapReduce

MapReduce框架把一些DIC應用的數據處理過程簡化抽象成map和reduce兩個階段，用戶在設計分布式程序時，只要實現Mapper和Reducer兩個函數，將數據分片，任務調度，集群容錯，集群間通信等問題，交由底層的MapReduce框架處理。

2.1 MapReduce體系架構

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

圖1 MapReduce架構示意圖

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

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

2.2 MapReduce DIC應用

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

DIC應用需要處理的數據具有以下幾個特點：

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

（2）數據形式是多樣的，往往不是真正的關系型數據（或稱為結構化數據），而可能是文本、網頁、XML文檔等無結構或者半結構化數據。

（3）數據分布是非集中式的，存儲于不同地理位置的多機架上。

MapReduce應用能自動完成數據密集型任務的分發(fā)和并行處理，如Facebook的服務事務日志分析、用戶操作分析、用戶點擊量分析等計算，Yahoo!部署Hadoop、PIG促進開源項日研究[1]。

3 分析與設計

本文的核心是在MapReduce模型框架中引入消息代理實體，管理控制Reduce-Worker傳來的接收消息；標記Mapper-Worker產生的中間結果，利用基于標簽的消息隊列實現異步傳輸模式，從而實現數據流的優(yōu)化。

3.1 消息代理機制

3.2 基于消息代理機制的MapReduce模型框架

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

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

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

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

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

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

圖3 MapReduce數據流對比示意圖

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

4 實驗測試

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

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

圖4 Sim-c1-MB的平均負載

圖5 Sim-m1-MB的平均負載

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

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

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

5 總結與展望

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

表1 硬件配置清單

[1]孫兆玉，袁志平，黃宇光.面向數據密集型計算Hadoop及其應用研究[C]//2008年全國高性能計算學術年會.無錫：[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的異步消息總線的設計與實現[J].計算機系統(tǒng)應用，2010，19（8）：254-257.

[6]李樂平，吳泉源.消息代理中間件InforBroker中集群技術的應用[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編程模型是廣泛應用于云計算環(huán)境下處理海量數據的一種并行計算框架。然而該框架下的面向數據密集型計算，集群節(jié)點間的數據傳輸依賴性較強，造成節(jié)點間的消息處理負載過重。提出基于消息代理機制的MapReduce改進模型，優(yōu)化數據流。經實驗數據表明，基于消息代理機制的MapReduce框架能提高數據密集型應用上的負載均衡。

消息代理；MapReduce；數據密集型計算；數據流

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—），男，博士，教授，主要研究領域為云計算、軟件工程；蔣仙（1987—），女，碩士研究生，主要研究領域為并行計算；方義秋（1963—），女，副教授，主要研究領域為中間件技術。E-mail：gejw@cqupt.edu.cn

2011-07-20

2012-01-13

1002-8331（2013）05-0120-03