五类实现框架对海量数据的统计

    1MapReduce介绍

　 　本文针对海量数据处理过程中的处理速度、存储空间、容错性、访问时间等方面存在的问题，通过对Google MapReduce编程模型的原理、执行流 程等进行分析研究，介绍5种主要的MapReduce实现框架：hadoop MapReduce、Apache Spark、Phoenix、 Disco、Mars，以期对MapReduce编程模型在行业内的使用前景有一个较全面的认识。

　　MapReduce介绍

　 　海量数据的处理对服务器CPU、I/O的吞吐都是严峻的考验，传统的技术架构和仅靠单台计算机基于串行的方式越来越不能适应当前海量数据处理的要求。只 有将这些计算进行并行化处理，通过提取出处理过程中存在的可并行工作的分量，用分布式模型来实现这些并行分量的并行执行过程。

　　MapReduce是由谷歌推出的一个编程模型，是一个能处理和生成超大规模数据集的算法模型，该架构能够在大量普通配置的计算机上实现并行化处理。

　 　MapReduce编程模型结合用户实现的Map和Reduce函数。用户自定义的Map函数处理一个输入的基于key/value pair的集合， 输出中间基于key/value pair的集合，MapReduce库把中间所有具有相同key值的value值集合在一起后传递给Reduce函数， 用户自定义的Reduce函数合并所有具有相同key值的value值，形成一个较小value值的集合。一般地，一个典型的MapReduce程序的执 行流程如图1所示。

　　MapReduce执行过程主要包括：

　　将输入的海量数据切片分给不同的机器处理；

　　执行Map任务的Worker将输入数据解析成key/value pair，用户定义的Map函数把输入的key/value pair转成中间形式的key/value pair；

　　按照key值对中间形式的key/value进行排序、聚合；

　　把不同的key值和相应的value集分配给不同的机器，完成Reduce运算；

　　任务成功完成后，MapReduce的输出存放在R个输出文件中，一般情况下，这R个输出文件不需要合并成一个文件，而是作为另外一个MapReduce的输入，或者在另一个可处理多个分割文件的分布式应用中使用。

　　MapReduce主要框架介绍

　 　Hadoop MapReduce是一个在计算机集群上分布式处理海量数据集的软件框架，包括一个JobTracker和一定数量的 TaskTracker。用户将MapReduce作业发送给Jobtracker所在集群的其他机器上分割工作，集群中其他机器执行 Tasktracker的Map或Reduce任务。

　　Spark是一个基于内存计算的开源的集群计算系统，目的是让数据分析更加快速。 Spark非常小巧玲珑，由加州伯克利大学AMP实验室的Matei为主的小团队所开发。使用的语言是Scala，项目的核心core部分的代码只有63 个Scala文件，非常短小精悍。Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。

　　 Phoenix作为斯坦福大学EE382a课程的一类项目，由斯坦福大学计算机系统实验室开发。Phoenix对MapReduce的实现原则和最初由 Google实现的MapReduce基本相同。不同的是，它在集群中以实现共享内存系统为目的，共享内存能最小化由任务派生和数据间的通信所造成的间接 成本。Phoenix可编程多核芯片或共享内存多核处理器(SMPs和ccNUMAs)，用于数据密集型任务处理。

　　Disco是由 Nokia研究中心开发的，基于MapReduce的分布式数据处理框架，核心部分由Erlang语言开发，外部编程接口为Python语言。Disco 是一个开放源代码的大规模数据分析平台，支持大数据集的并行计算，能运行在不可靠的集群计算机上。Disco可部署在集群和多核计算机上，还可部署在 Amazon EC2 上。

　　Mars是香港科技大学与微软、新浪合作开发的基于GPU的MapReduce框架。目前已经包含字符串匹配、矩阵乘法、倒排索引、字词统计、网页访问排名、网页访问计数、相似性评估和K均值等8项应用，能够在32位与64位的Linux平台上运行。

　　针对5种框架的特点笔者进行了如下分类。

　　编程语言

　　Hadoop MapReduce： Hadoop采用Java开发，所以能很好地支持Java语言编写的MapReduce作业，如果采用C/C++或其他语言编写MapReduce作业，需要用到Hadoop Streaming或Hadoop Pipes工具；

　　Spark：Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地集合对象一样轻松地操作分布式数据集；

　　Phoenix：采用全C++编写，总代码量不超过1万行，提供C和C++的应用程序接口；

　　Disco：核心部分采用并发性能很高的Erlang语言开发，其外部编程接口为易于编程的Python语言；

　　Mars：采用C++编写，提供C、C++的应用程序编程接口，支持最新的CUDA SDK。

　　构建平台

　　Hadoop MapReduce：需要首先架构基于Hadoop的集群系统，通过HDFS完成运算的数据存储工作；

　　Spark：可以的单独运行，也可以与Hadoop框架完整结合；

　　Phoenix：独立运行，不需要提前部署集群，运行时系统的实现是建立在PThread之上的，也可方便地移植到其他共享内存线程库上；

　　Disco：整个Disco平台由分布式存储系统DDFS和MapReduce框架组成，DDFS与计算框架高度耦合，通过监控各个节点上的磁盘使用情况进行负载均衡；

　　Mars：运行时为Map或Reduce任务初始化大量的GPU线程，并为每个线程自动分配少量的key/value对来运行任务。

　　功能特点

　　Hadoop MapReduce：计算能力非常强，适合超大数据集的应用程序，但是由于系统开销等原因，处理小规模数据的速度不一定比串行程序快，并且本身集群的稳定性不高；

　　Spark：在保证容错的前提下，用内存来承载工作集，内存的存取速度快于磁盘多个数量级，从而可以极大提升性能；

　　Phoenix：利用共享内存缓冲区实现通信，从而避免了因数据复制产生的开销，但Phoenix也存在不能自动执行迭代计算、没有高效的错误发现机制等不足；

　　Disco：由一个Master服务器和一系列Worker节点组成，Master和Worker之间采用基于轮询的通信机制，通过HTTP的方式传输数据。轮询的时间间隔不好确定，若时间间隔设置不当，会显著降低程序的执行性能；

　 　Mars：由于GPU线程不支持运行时动态调度，所以给每个GPU线程分配的任务是固定的，若输入数据划分布均匀，将导致Map或Reduce阶段的负 载不均衡，使得整个系统性能急剧降低。同时由于GPU不支持运行时在设备内存中分配空间，需要预先在设备内存中分配好输入数据和输出数据的存放空间，但是 Map和Reduce阶段输出数据大小是未知的，并且当多个GPU线程同时向共享输出区域中写数据时，易造成写冲突。

　　五类实现框架对海量文本数据的统计实验

　　单词计数(WordCount)是最简单也是最能体现MapReduce思想的程序之一，可以称为MapReduce版“Hello World”。单词计数主要完成功能是：统计一系列文本文件中每个单词出现的次数。

　　WordCount的实现步骤：

　　1、将文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成对，如图2-1所示。这一步由MapReduce框架自动完成，其中偏移量(即key值)包括了回车所占的字符数(Windows和Linux环境会不同)。

　　2、将分割好的对交给用户定义的map方法进行处理，生成新的对。

　　3、得到map方法输出的对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。

　　4、Reducer先对从Mapper接收的数据进行排序，再交由用户自定义的reduce方法进行处理，得到新的对，并作为WordCount的输出结果。

　　本次实验的硬件资源基于x86服务器1台，配置内存为32GB DDR3，E5 CPU/12 核，GPU。实验数据样本为10M/50M/100M/500M/1000M的文本文件五个，我们使用Hadoop MapReduce、Spark、 Phoenix、Disco、Mars等MapReduce框架分别运行文本分析程序，基于结果一致的前提下统计出运行时间、运行时CPU占有率、运行时 内存占有率等数据，并采用这些数据绘制成柱状图。

　　Hadoop MapReduce的运行时间最长，原因是Hadoop生态环境包含内 容过多，所以每次任务启动时首先需要加载所需资源包，然后缓慢地发起任务，并且由于本身是用性能较差的Java语言编写的，所以导致整体计算时间长、性能 差。Phoenix由于采用汇编和C语言编写，内核很小，运行时所用资源很少，所以整个测试过程耗时也较少。Mars由于必须在GPU上运行，本身GPU 由于价格因素，导致不太可能在实际应用场景里推广，所以Phoenix的性价比是最高的。需要时长从高到低分别是Hadoop MapReduce、 Disco、Spark、Phoenix、Mars。

　　Hadoop MapReduce、Disco这两个框架需要占用的CPU资源在 1000M文本处理时基本到达最大饱和度(大于90%)，Apache Spark的CPU使用率没有完全伴随着文本文件增大而大幅上涨，Phoenix 和Mars基本控制在性价比较高的范围内。

　　Mars和Phoenix使用的内存在数据量较小时是最少的，Apache Spark为随着数据量增大而大幅增加，在数据量最大时它对内存的消耗是最小的。Hadoop MapReduce和Disco都需要占用较多的内存。

　 　从上面的测试结果我们得出，如果用户只需要处理海量的文本文件，不需要考虑存储、二次数据挖掘等，采用Phoenix是最大性价比的选择。如果应用程序 需要处理的数据量非常大，并且客户希望计算出的数据可以被存储和二次计算或数据挖掘，那Hadoop MapReduce较好，因为整个Hadoop生态 圈庞大，支持很好。Apache Spark由于架构层面设计不同，所以对于CPU、内存的使用率一直保持较低状态，它未来可以用于海量视频分析用途。

　　五类实现框架结合视频人脸分析的实验

　 　安防行业的并行测试实验大多是基于智能视频分析技术基础之上的。智能视频分析技术是一种基于人工智能的识别模式。它综合了各种高科技研究成果，主要借助 智能视频分析技术的处理方法，在结合一些硬件设施，对某些对象(比如人员、车辆等)进行研究和处理，形成一种核心算法。

　　在本次测试中，我们针对的是人脸特征抓取实验，即通过对一段指定录像分析，提取出录像中所有出现的人脸图片的过程。对录像中出现的人脸图片分析过程大致上可以分为三个阶段：取流、解码及分析、提取物发送。码流分析提取服务即我们本次实验所需要的三个阶段。

　　主计算节点把录像文件读入到内存中，将码流分割为若干个子块分发给从计算节点。由于我们采用的是MapReduce框架，所以程序会自动分为若干个线程执行，每个线程对应一个Map，每个Map都会执行解码、分析、结果输出三个步骤。

　 　我们在x86机器上进行了本次实验，实验数据是一个2.66GB大小的包含1092个人脸的录像文件，录像分辨率为1080P。我们通过分别采用不同的 MapReduce框架来运行程序，对程序运行结果进行匹配，5个框架的运行结果完全一致，即抓取出1092个人脸图片。我们对程序运行时间、运行过程中 CPU使用率、运行过程中内存使用率做了统计并生成柱状图供参考。本次实验过程中所使用的人脸检测算法是笔者公司图像处理与智能分析部门自主研发的算法。

　 　Mars和Phoenix框架处理录像所需时间最短，运行智能分析程序时CPU使用率对于所有框架基本上都达到最大饱和度(90%以上)。由于所做的实 验是对码流进行分析，码流本身需要占用较大的内存空间，解码、分析等处理过程也许要占用内存用于存放中间结果，所以内存基本上也达到最大使用饱和度 (90%以上)。综上所述，CPU和内存的使用率在本类实验过程中不需要过多考虑，最主要的对比点是运行时间。考虑到Mars必须基于GPU运行，并且 GPU的价格较高，所以Phoenix的性价比更高。Hadoop MapReduce虽然处理时间最长，但是它具有强大的生态环境，利于对处理结果数据 进行保存和数据挖掘，所以对于大型公司来说它依然是很好的选择。Apache Spark虽然在本次实验中没有太多亮点，但是从各类大数据学术会议上得到 的反馈较好，它基于内存方式的运算模式可以帮助处理海量数据，未来一定可以在智能分析领域有很大的作为。

　　结语

　　现实世 界很多实例都可用MapReduce编程模型来表示，MapReduce作为一个通用可扩展的、高容错性的并行处理模型，可有效地处理海量数据，不断地从 中分析挖掘出有价值的信息。MapReduce封装了并行处理、负载均衡、容错、数据本化等技术难点细节。通过本文的两例测试用例可以证明 MapReduce 适用于海量文本分析、海量视频智能分析等安防行业密切相关的应用场景，诸如行为分析、车牌识别、人脸抓拍、客流统计等智能化技术的应 用，尤其是对海量视频执行高并发处理，可以很好地在平安城市、智慧城市等大型安防项目中落地，为公安机关治安管理、案件侦破等提供有力的技术支持。