MapReduce迭代式计算发展与应用

马在营　刘建新　徐彬

摘要摘要：MapReduce的优势集中体现在并行计算上，而在迭代计算上则存在诸多不足。科研人员不断对MapReduce并行计算模型进行迭代计算优化，使MapReduce可以支持显示迭代式计算。介绍了传统MapReduce框架与迭代式MapReduce框架，通过K-means算法测试了iMapReduce、Hadoop MapReduce的迭代性能，给出了实验结果及分析。

关键词关键词：迭代式计算；MapReduce；Haloop；Hadoop；iMapReduce

DOIDOI：10.11907/rjdk.162775

中图分类号：TP301

文献标识码：A文章编号文章编号：16727800（2017）005020703

0引言

在全球信息产业高速发展融合的背景下，网络数据资源规模急剧膨胀，尤其是高能物理、互联网应用、基因工程、电子商务以及计算机仿真等领域的数据量攀升速度惊人，现有数据分析工具很难满足日益增长的海量密集型数据的信息处理需求。基于此，Google实验室专门设计了MapReduce编程模型，该模型在数据信息的批量处理上优势明显，但在迭代处理上问题也相当突出。逐次逼近是迭代计算的基本思想，迭代计算过程为先取近似值，然后采用递推公式对近似值反复校正，直至精度达到要求为止。当数据量较小时，可以在单机上进行迭代计算，但当数据量非常大时，迭代处理则极其耗时。在数据挖掘、信息检索、机器学习等领域，有很多算法需要迭代，传统的迭代计算不能有效应对当前的大数据处理需求。经过几年时间，科研工作者对MapReduce进行迭代计算改进，产生了一些迭代式计算框架，包括比较知名的Twister、Haloop等。

1传统MapReduce模型框架

Google于2004年在论文《分布式计算：基于大型集群的数据简化处理》中首次提出MapReduce框架，指出MapReduce框架在处理密集型应用数据过程中将处理程序简化抽象为Map与Reduce两个阶段，用户在进行分布式程序设计过程中，只需调用reduce（）和map（）两个函数即可，无需过多考虑任务调度、设备通信、数据分片以及容错等细节问题。在MapReduce框架内，这些问题都能得到很好的处理。作为MapReduce框架的主要思想，Map与Reduce来自函数编程语言，其原理如图1所示。Map将数据打散，Reduce则负责聚集这些数据。在Map环节，maptask每读取一个block，即采用map（）函数对数据进行处理，同时将处理结果写入本地磁盘；在Reduce环节，每个reduce task在对Map Task节点数据进行远程读取过程中，都会采用reduce（）函数处理数据，同时将处理完成的数据写入分布式文件系统内。在MapReduce框架内，用户只要通过map和reduce端口，即可快速计算TB级数据，如数据挖掘和日志分析等常见应用。此外，MapReduce框架还适用于圆周率等科学数据的计算。

通过上述分析可知，在传统MapReduce框架内，无论是map环节，还是reduce环节，数据处理结果均需写入磁盘内，虽然这一过程会降低系统性能，但是能够提高系统可靠性。也正因如此，传统MapReduce在迭代计算处理上问题突出，若用户强行在传统MapReduce上进行迭代计算，系统性能则会变得非常差。

2迭代式MapReduce框架

2.1Twister

在Twister内，大文件不会被自动切割为单个block，所以用户必须提前将文件分割为小文件才能进行task处理。在map环节，调用map（）函数处理后，数据结果存储于分布式内存之中，然后利用broker network将数据推送至reduce task（若Twister内存较大，则所有中间数据都能存储其中）；在reduce环节，通过combine对reducetask产生的结果进行归并，此时用户可依据条件作出是否结束迭代计算的决定。合并后的数据被分解传送至map task，进行新一轮迭代计算。为有效提高系统容错性，Twister会定时将reducetask和maptask产生的结果录入磁盘内，避免task失败后数据丢失。即使task失败，依然可以从磁盘内调取数据重新进行迭代处理。Twister架构如图2所示。

为避免用户在迭代计算中对task的重建，Twister建立了一个taskpool，当用户需要使用task時，可直接从pool中读取。在Twister内，所有数据与消息均由broker network传递，brokernetwork是独立模块，现阶段仅支持ActiveMQ和NaradaBroking。目前Twister尚属于研究性项目，其设计策略决定了Twister还不能在实际中得到广泛应用。例如Twister将数据存储于分布式内存中，而缺乏分布式文件系统，仅提供tool对文件进行访问与存储，并存在迭代计算模型不够抽象、支持应用类型偏少等问题。

2.2Haloop

Haloop是Haloop Mapreduce 的修改版，Haloop不仅扩展了分布迭代编程，并通过使用任务调度器loopaware添加各种缓存机制，大大提高了效率。其架构如图3所示。

图3有3个工作在运行，工作1、工作2、工作3。每个工作都有3个任务同时从slave节点上运行。为了适应迭代计算，Haloop以Hadoop为基础作了一些调整：①Haloop提供了一个新的应用程序用户编程接口，简化了迭代表达式的表达；②Haloop的master节点包含一个新的loop控制模块，可以指定迭代停止条件；③Haloop为迭代计算使用一个新的任务调度器，可以将数据本地化；④Haloop的缓存和索引应用程序在slave节点上。

（1）在Haloop内迭代式任务全部被抽象为：

Ro代表初始输入，Ri代表第i次迭代的结果，L是迭代计算中保持不变的数据。Haloop主要编程接口为：

SetFixedPointThreshold：设置好迭代计算的结束条件，即距离差的阈值。

Set the number of iterations：设置迭代次数。

Setiterationinput：设置迭代变化输入数据。

AddinvariantTable：设置不变的输入数据。

（2） Loop-aware 任务调度。 Haloop 在初次迭代计算中会将不变的输入数据存储于计算节点中，以便后续的task调度，且数据应当尽量存储于locality等固定节点中，使每次迭代计算时无需重新传输数据。

（3）Index与Cache。无论是map task的輸入还是输出，reducetask的输出都会通过缓存与建索引来加快迭代计算速度。其中，缓存主要指迭代计算结果被写入磁盘以供循环迭代使用。

总体来看，与Twister相比，Haloop更为抽象，能够支持多种计算。此外，Haloop是在Hadoop基础上优化改进而成，因此具备Hadoop的优点。

2.3iMapReduce

张岩峰在论文《iMapReduce：分布式计算框架迭代计算》中提出iMapReduce概念，iMapReduce是以尽量减少系统开销为目的而构建的高效迭代计算框架。MapReduce需要一系列的迭代作业来处理迭代计算，图4左图是MapReduce迭代处理过程。在iMapReduce中，迭代任务只发生在初始阶段和结束阶段，用户仅需建立一个作业任务即能完成迭代计算，避免了对系统的反复开销。通过对本地静态数据的维护，减少因反复加载数据而产生的系统开销，同时允许在单次迭代计算中异步执行map任务。iMapReduce极大地提高了迭代算法性能。图4展示了iMapReduce的迭代计算过程。iMapReduce为用户提供编程的API接口。iMapReduce兼容Hadoop MapReduce任务，用户可根据实际决定是否启用iMapReduce功能。iMapReduce将静态数据和迭代数据区分开。图5是iMapReduce迭代计算节点数据流，在初始环节中，静态数据与迭代数据被加载至本地文件系统，其中静态数据始终存储于本地文件系统内，而迭代数据在调用map（）函数处理前，需要同静态数据联合执行操作，即将静态数据与迭代数据连接起来，作为map（）函数输入。

通过这些迭代优化，使iMapReduce更加适合大数据迭代处理，并且提供了编程接口API，兼容HadoopMapreduce任务，用户可选择是否启用iMapReduce的迭代处理功能。

3MapReduce迭代计算迭代性能测试

3.1实验设计

集群环境由5台计算机组成（1个Master节点，4个Slave节点），系统为64位Windows7，处理器为i7-4790 3.60GHz，内存8GB，网速10Mb/s。

K-means 算法是聚类分析中使用最广泛的算法之一，可用于测试不同平台的迭代性能。已知观测集（x1，x2，x3，...，xn），其中每个观测都是一个d维实向量，k-平均聚类要将这n个观测划分到k个集合中（k≤n），使组内平方和（WCSS，Within-Cluster Sum of Squares）最小。换言之，它的目标是找到使式（1）满足的聚类Si，其中μi是Si中所有点的均值。简单描述为：不断迭代计算各个数据簇的中心点，直到该中心点趋于稳定。

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

K-means步骤如下：①创建k个数据簇的中心点；②计算所有数据点到k个中心点的距离，将其划归到距离自己最近的中心点；③根据上次聚类结果，计算各个数据簇的算数平均值作为新的数据簇中心点；④将所有数据在新中心点上重新聚类；⑤重复第4步，直到中心点趋于稳定。

通过测试K-means算法在Hadoop Mapreduce、iMapReduce的运行情况，统计出不同平台运行K-means的时间。

3.2实验结果

从表1中可以看出，iMapReduce运行时间远远短于Hadoop MapReduce，大约是Hadoop MapReduce运行时间的25%～39%，这是由于Hadoop不支持迭代计算，花了大量时间在数据通信上，从而大大降低了执行速度。而iMapReduce通过避免传输静态数据而减少了通信开销，并在一次迭代内允许异步执行，因此对于迭代应用有良好的性能表现。

4结语

迭代式计算是大数据领域中一种重要的计算方法，Twister和Haloop的模型抽象度不够高，支持的计算有限，iMapReduce有效提升了大数据迭代计算的性能。目前MapReduce迭代式计算还处于发展阶段，在以后的研究中，如何优化提高分布式迭代计算的作业调度速度是亟待解决的问题。

参考文献参考文献：

[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]张岩峰.云环境下大数据迭代计算研究[J].沈阳：东北大学，2012.

[5]易修文， 李天瑞， 张钧波，等. 不同MapReduce运行系统的性能测试与分析[J].计算机科学， 2015，42（5）：2427.

[6]董的博客. 传统MapReduce框架介绍[EB/OL]. http：//dongxicheng.org/mapreduce/traditionalmapreduceframework/.

[7]Kmeans算法的Hadoop实现[EB/OL].http：//blog.csdn.net/chaaaa_wangyc/article/details/53426612？locationNum=9&fps=1.

责任编辑（责任编辑：黄健）