五种基于 MapReduce 的并行计算框架介绍及性能测试（2）MapReduce

look_w

MapReduceMapReduce 是由谷歌推出的一个编程模型，是一个能处理和生成超大数据集的算法模型，该架构能够在大量普通配置的计算机上实现并行化处理。MapReduce                编程模型结合用户实现的 Map 和 Reduce 函数。用户自定义的 Map 函数处理一个输入的基于 key/value pair 的集合，输出中间基于                key/value pair 的集合，MapReduce 库把中间所有具有相同 key 值的 value 值集合在一起后传递给 Reduce 函数，用户自定义的                Reduce 函数合并所有具有相同 key 值的 value 值，形成一个较小 value 值的集合。一般地，一个典型的 MapReduce 程序的执行流程如图 1                所示。
图 1 .MapReduce                    程序执行流程图MapReduce 执行过程主要包括：

• 将输入的海量数据切片分给不同的机器处理；
• 执行 Map 任务的 Worker 将输入数据解析成 key/value pair，用户定义的 Map 函数把输入的 key/value pair                    转成中间形式的 key/value pair；
• 按照 key 值对中间形式的 key/value 进行排序、聚合；
• 把不同的 key 值和相应的 value 集分配给不同的机器，完成 Reduce 运算；
• 输出 Reduce 结果。

任务成功完成后，MapReduce 的输出存放在 R 个输出文件中，一般情况下，这 R 个输出文件不需要合并成一个文件，而是作为另外一个 MapReduce                的输入，或者在另一个可处理多个分割文件的分布式应用中使用。
受 Google MapReduce 启发，许多研究者在不同的实验平台上实现了 MapReduce 框架，本文将对 Apache Hadoop                MapReduce、Apache、Spark、斯坦福大学的 Phoenix，Nokia 研发的 Disco，以及香港科技大学的 Mars 等 5 个 MapReduce                实现框架进行逐一介绍和各方面对比。

• Hadoop MapReduce

Hadoop 的设计思路来源于 Google 的 GFS 和                MapReduce。它是一个开源软件框架，通过在集群计算机中使用简单的编程模型，可编写和运行分布式应用程序处理大规模数据。Hadoop 包含三个子项目：Hadoop                Common、Hadoop Distributed File System(HDFS) 和 Hadoop MapReduce。
第一代 Hadoop MapReduce 是一个在计算机集群上分布式处理海量数据集的软件框架，包括一个 JobTracker 和一定数量的                TaskTracker。运行流程图如图 2 所示。
图 2 .Hadoop                    MapReduce 系统架构图在最上层有 4 个独立的实体，即客户端、JobTracker、TaskTracker 和分布式文件系统。客户端提交 MapReduce 作业；JobTracker                协调作业的运行；JobTracker 是一个 Java 应用程序，它的主类是 JobTracker；TaskTracker 运行作业划分后的任务，TaskTracker                也是一个 Java 应用程序，它的主类是 TaskTracker。Hadoop 运行 MapReduce                作业的步骤主要包括提交作业、初始化作业、分配任务、执行任务、更新进度和状态、完成作业等 6 个步骤。

• Spark MapReduce

Spark 是一个基于内存计算的开源的集群计算系统，目的是让数据分析更加快速。Spark 非常小巧玲珑，由加州伯克利大学 AMP 实验室的 Matei                为主的小团队所开发。使用的语言是 Scala，项目的核心部分的代码只有 63 个 Scala 文件，非常短小精悍。Spark                启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。Spark                提供了基于内存的计算集群，在分析数据时将数据导入内存以实现快速查询，“速度比”基于磁盘的系统，如比 Hadoop 快很多。Spark                最初是为了处理迭代算法，如机器学习、图挖掘算法等，以及交互式数据挖掘算法而开发的。在这两种场景下，Spark 的运行速度可以达到 Hadoop 的几百倍。

• Disco

Disco 是由 Nokia 研究中心开发的，基于 MapReduce 的分布式数据处理框架，核心部分由 Erlang 语言开发，外部编程接口为 Python                语言。Disco 是一个开放源代码的大规模数据分析平台，支持大数据集的并行计算，能运行在不可靠的集群计算机上。Disco 可部署在集群和多核计算机上，还可部署在                Amazon EC2 上。Disco 基于主/从架构 (Master/Slave)，图 3 总体设计架构图展示了通过一台主节点 (Master) 服务器控制多台从节点                (Slave) 服务器的总体设计架构。
图 3 .Disco                    总体架构图Disco 运行 MapReduce 步骤如下：

• Disco 用户使用 Python 脚本开始 Disco 作业；
• 作业请求通过 HTTP 发送到主机；
• 主机是一个 Erlang 进程，通过 HTTP 接收作业请求；
• 主机通过 SSH 启动每个节点处的从机；
• 从机在 Worker 进程中运行 Disco 任务。

• Phoenix

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

• Mars

Mars 是香港科技大学与微软、新浪合作开发的基于 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 个操作步骤：

• 在主存储器中输入 key/value 对，并将它们存储到数组；
• 初始化运行时的配置参数；
• 复制主存储器中的输入数组到 GPU 设备内存；
• 启动 GPU 上的 Map 阶段，并将中间的 key/value 对存储到数组；
• 如果 mSort 选择 F，即需要排序阶段，则对中间结果进行排序；
• 如果 noReduce 是 F，即需要 Reduce 阶段，则启动 GPU 上的 Reduce 阶段，并输出最终结果，否则中间结果就是最终结果；
• 复制 GPU 设备存储器中的结果到主存储器。

上述步骤的 1，2，3，7 这四个步骤的操作由调度器来完成，调度器负责准备数据输入，在 GPU 上调用 Map 和 Reduce 阶段，并将结果返回给用户。
五种框架的优缺点比较
表 1. 五种框架优缺点比较
Hadoop MapReduce  Spark  Phoenix  Disco  Mars  编程语言  Java 为主  Scala  C  Erlang  C++  构建平台  需要首先架构基于 Hadoop 的集群系统，通过 HDFS 完成运算的数据存储工作  可以的单独运行，也可以与 Hadoop 框架完整结合  独立运行，不需要提前部署集群，运行时系统的实现是建立在 PThread 之上的，也可方便地移植到其他共享内存线程库上  整个 Disco 平台由分布式存储系统 DDFS 和 MapReduce 框架组成，DDFS                            与计算框架高度耦合，通过监控各个节点上的磁盘使用情况进行负载均衡  运行时为 Map 或 Reduce 任务初始化大量的 GPU 线程，并为每个线程自动分配少量的 key/value 对来运行任务  功能特点  计算能力非常强，适合超大数据集的应用程序，但是由于系统开销等原因，处理小规模数据的速度不一定比串行程序快，并且本身集群的稳定性不高  在保证容错的前提下，用内存来承载工作集，内存的存取速度快于磁盘多个数量级，从而可以极大提升性能  利用共享内存缓冲区实现通信，从而避免了因数据复制产生的开销，但 Phoenix 也存在不能自动执行迭代计算、没有高效的错误发现机制等不足  由一个 Master 服务器和一系列 Worker 节点组成，Master 和 Worker 之间采用基于轮询的通信机制，通过 HTTP                            的方式传输数据。轮询的时间间隔不好确定，若时间间隔设置不当，会显著降低程序的执行性能  由于 GPU 线程不支持运行时动态调度，所以给每个 GPU 线程分配的任务是固定的，若输入数据划分布均匀，将导致 Map 或 Reduce                            阶段的负载不均衡，使得整个系统性能急剧降低。同时由于 GPU                            不支持运行时在设备内存中分配空间，需要预先在设备内存中分配好输入数据和输出数据的存放空间，但是 Map 和 Reduce                            阶段输出数据大小是未知的，并且当多个 GPU 线程同时向共享输出区域中写数据时，易造成写冲突