大讲台分享五种基于MapReduce的并行计算框架介绍及性能测试

大讲台分享：五种基于MapReduce的并行计算框架介绍及性能测试当使用Hadoop技术架构集群，集群内新增、删除节点，或者某个节点机器内硬盘存储达到饱和值时，都会造成集群内数据分布不均匀、数据丢失风险增加等问题出现。本文对HDFS内部的数据平衡方式做了介绍，通过实验案例的方式向读者解释内部数据平衡的解决办法。并行计算模型和框架目前开源社区有许多并行计算模型和框架可供选择，按照实现方式、运行机制、依附的产品生态圈等可以被划分为几个类型，每个类型各有优缺点，如果能够对各类型的并行计算框架都进行深入研究及适当的缺点修复，就可以为不同硬件环境下的海量数据分析需求提供不同的软件层面的解决方案。并行计算框架并行计算或称平行计算是相对于串行计算来说的。它是一种一次可执行多个指令的算法，目的是提高计算速度，以及通过扩大问题求解规模，解决大型而复杂的计算问题。所谓并行计算可分为时间上的并行和空间上的并行。时间上的并行就是指流水线技术，而空间上的并行则是指用多个处理器并发的执行计算。并行计算（ParallelComputing）是指同时使用多种计算资源解决计算问题的过程，是提高计算机系统计算速度和处理能力的一种有效手段。它的基本思想是用多个处理器来协同求解同一问题，即将被求解的问题分解成若干个部分，各部分均由一个独立的处理机来并行计算。并行计算系统既可以是专门设计的、含有多个处理器的超级计算机，也可以是以某种方式互连的若干台的独立计算机构成的集群。通过并行计算集群完成数据的处理，再将处理的结果返回给用户。国内外研究欧美发达国家对于并行计算技术的研究要远远早于我国，从最初的并行计算逐渐过渡到网格计算，随着Internet网络资源的迅速膨胀，因特网容纳了海量的各种类型的数据和信息。海量数据的处理对服务器CPU、IO的吞吐都是严峻的考验，不论是处理速度、存储空间、容错性，还是在访问速度等方面，传统的技术架构和仅靠单台计算机基于串行的方式越来越不适应当前海量数据处理的要求。国内外学者提出很多海量数据处理方法，以改善海量数据处理存在的诸多问题。目前已有的海量数据处理方法在概念上较容易理解，然而由于数据量巨大，要在可接受的时间内完成相应的处理，只有将这些计算进行并行化处理，通过提取出处理过程中存在的可并行工作的分量，用分布式模型来实现这些并行分量的并行执行过程。随着技术的发展，单机的性能有了突飞猛进的发展变化，尤其是内存和处理器等硬件技术，但是硬件技术的发展在理论上总是有限度的，如果说硬件的发展在纵向上提高了系统的性能，那么并行技术的发展就是从横向上拓展了处理的方式。2003年美国Google公司对外发布了MapReduce、GFS、BigData三篇论文，至此正式将并行计算框架落地为MapReduce框架。我国的并行和分布式计算技术研究起源于60年代末，按照国防科技大学周兴铭院士提出的观点，到目前为止已经三个阶段了。第一阶段，自60年代末至70年代末，主要从事大型机内的并行处理技术研究；第二阶段，自70年代末至90年代初，主要从事向量机和并行多处理器系统研究；第三阶段，自80年代末至今，主要从事MPP(MassivelyParallelProcessor)系统研究。尽管我国在并行计算方面开展的研究和应用较早，目前也拥有很多的并行计算资源，但研究和应用的成效相对美国还存在较大的差距，有待进一步的提高和发展。MapReduceMapReduce是由谷歌推出的一个编程模型，是一个能处理和生成超大数据集的算法模型，该架构能够在大量普通配置的计算机上实现并行化处理。MapReduce编程模型结合用户实现的Map和Reduce函数。用户自定义的Map函数处理一个输入的基于key/valuepair的集合，输出中间基于key/valuepair的集合，MapReduce库把中间所有具有相同key值的value值集合在一起后传递给Reduce函数，用户自定义的Reduce函数合并所有具有相同key值的value值，形成一个较小value值的集合。一般地，一个典型的MapReduce程序的执行流程如图1所示。图1.MapReduce程序执行流程图MapReduce执行过程主要包括：1.将输入的海量数据切片分给不同的机器处理；2.执行Map任务的Worker将输入数据解析成key/valuepair，用户定义的Map函数把输入的key/valuepair转成中间形式的key/valuepair；3.按照key值对中间形式的key/value进行排序、聚合；4.把不同的key值和相应的value集分配给不同的机器，完成Reduce运算；5.输出Reduce结果。任务成功完成后，MapReduce的输出存放在R个输出文件中，一般情况下，这R个输出文件不需要合并成一个文件，而是作为另外一个MapReduce的输入，或者在另一个可处理多个分割文件的分布式应用中使用。受GoogleMapReduce启发，许多研究者在不同的实验平台上实现了MapReduce框架，本文将对ApacheHadoopMapReduce、Apache、Spark、斯坦福大学的Phoenix，Nokia研发的Disco，以及香港科技大学的Mars等5个MapReduce实现框架进行逐一介绍和各方面对比。HadoopMapReduceHadoop的设计思路来源于Google的GFS和MapReduce。它是一个开源软件框架，通过在集群计算机中使用简单的编程模型，可编写和运行分布式应用程序处理大规模数据。Hadoop包含三个子项目：HadoopCommon、HadoopDistributedFileSystem(HDFS)和HadoopMapReduce。第一代HadoopMapReduce是一个在计算机集群上分布式处理海量数据集的软件框架，包括一个JobTracker和一定数量的TaskTracker。运行流程图如图2所示。图2.HadoopMapReduce系统架构图在最上层有4个独立的实体，即客户端、JobTracker、TaskTracker和分布式文件系统。客户端提交MapReduce作业；JobTracker协调作业的运行；JobTracker是一个Java应用程序，它的主类是JobTracker；TaskTracker运行作业划分后的任务，TaskTracker也是一个Java应用程序，它的主类是TaskTracker。Hadoop运行MapReduce作业的步骤主要包括提交作业、初始化作业、分配任务、执行任务、更新进度和状态、完成作业等6个步骤。SparkMapReduceSpark是一个基于内存计算的开源的集群计算系统，目的是让数据分析更加快速。Spark非常小巧玲珑，由加州伯克利大学AMP实验室的Matei为主的小团队所开发。使用的语言是Scala，项目的核心部分的代码只有63个Scala文件，非常短小精悍。Spark启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。Spark提供了基于内存的计算集群，在分析数据时将数据导入内存以实现快速查询，“速度比”基于磁盘的系统，如比Hadoop快很多。Spark最初是为了处理迭代算法，如机器学习、图挖掘算法等，以及交互式数据挖掘算法而开发的。在这两种场景下，Spark的运行速度可以达到Hadoop的几百倍。DiscoDisco是由Nokia研究中心开发的，基于MapReduce的分布式数据处理框架，核心部分由Erlang语言开发，外部编程接口为Python语言。Disco是一个开放源代码的大规模数据分析平台，支持大数据集的并行计算，能运行在不可靠的集群计算机上。Disco可部署在集群和多核计算机上，还可部署在AmazonEC2上。Disco基于主/从架构(Master/Slave)，图3总体设计架构图展示了通过一台主节点(Master)服务器控制多台从节点(Slave)服务器的总体设计架构。图3.Disco总体架构图Disco运行MapReduce步骤如下：1.Disco用户使用Python脚本开始Disco作业；2.作业请求通过HTTP发送到主机；3.主机是一个Erlang进程，通过HTTP接收作业请求；4.主机通过SSH启动每个节点处的从机；5.从机在Worker进程中运行Disco任务。PhoenixPhoenix作为斯坦福大学EE382a课程的一类项目，由斯坦福大学计算机系统实验室开发。Phoenix对MapReduce的实现原则和最初由Google实现的MapReduce基本相同。不同的是，它在集群中以实现共享内存系统为目的，共享内存能最小化由任务派生和数据间的通信所造成的间接成本。Phoenix可编程多核芯片或共享内存多核处理器(SMPs和ccNUMAs)，用于数据密集型任务处理。MarsMars是香港科技大学与微软、新浪合作开发的基于GPU的MapReduce框架。目前已经包含字符串匹配、矩阵乘法、倒排索引、字词统计、网页访问排名、网页访问计数、相似性评估和K均值等8项应用，能够在32位与64位的Linux平台上运行。Mars框架实现方式和基于CPU的MapReduce框架非常类似，也由Map和Reduce两个阶段组成，它的基本工作流程图如图4所示。图4.Mars基本工作流程图在开始每个阶段之前，Mars初始化线程配置，包括GPU上线程组的数量和每个线程组中线程的数量。Mars在GPU内使用大量的线程，在运行时阶段会均匀分配任务给线程，每个线程负责一个Map或Reduce任务，以小数量的key/value对作为输入，并通过一种无锁的方案来管理MapReduce框架中的并发写入。Mars的工作流程主要有7个操作步骤：1.在主存储器中输入key/value对，并将它们存储到数组；2.初始化运行时的配置参数；3.复制主存储器中的输入数组到GPU设备内存；4.启动GPU上的Map阶段，并将中间的key/value对存储到数组；5.如果mSort选择F，即需要排序阶段，则对中间结果进行排序；6.如果noReduce是F，即需要Reduce阶段，则启动GPU上的Reduce阶段，并输出最终结果，否则中间结果就是最终结果；7.复制GPU设备存储器中的结果到主存储器。上述步骤的1，2，3，7这四个步骤的操作由调度器来完成，调度器负责准备数据输入，在GPU上调用Map和Reduce阶段，并将结果返回给用户。五种框架的优缺点比较表1.五种框架优缺点比较HadoopMapReduceSparkPhoenixDiscoMars编程语言Java为主ScalaCErlangC++构建平台需要首先架构基于Hadoop的集群系统，通过HDFS完成运算的数据存储工作可以的单独运行，也可以与Hadoop框架完整结合独立运行，不需要提前部署集群，运行时系统的实现是建立在PThread之上的，也可方便地移植到其他共享内存线程库上整个Disco平台由分布式存储系统DDFS和MapReduce框架组成，DDFS与计算框架高度耦合，通过监控各个节点上的磁盘使用情况进行负载均衡运行时为Map或Reduce任务初始化大量的GPU线程，并为每个线程自动分配少量的key/value对来运行任务功能特点计算能力非常强，适合超大数据集的应用程序，但是由于系统开销等原因，处理小规模数据的速度不一定比串行程序快，并且本身集群的稳定性不高在保证容错的前提下，用内存来承载工作集，内存的存取速度快于磁盘多个数量级，从而可以极大提升性能利用共享内存缓冲区实现通信，从而避免了因数据复制产生的开销，但Phoenix也存在不能自动执行迭代计算、没有高效的错误发现机制等不足由一个Master服务器和一系列Worker节点组成，Master和Worker之间采用基于轮询的通信机制，通过HTTP的方式传输数据。轮询的时间间隔不好确定，若时间间隔设置不当，会显著降低程序的执行性能由于GPU线程不支持运行时动态调度，所以给每个GPU线程分配的任务是固定的，若输入数据划分