3.4 MapReduce概述

3.4.1 MapReduce的概念

在云计算和大数据技术领域被广泛提到并被成功应用的一项技术是MapReduce。MapReduce是谷歌系统和Hadoop系统中的一项核心技术。它是一个软件框架，可以将单个计算作业分配给多台计算机执行。它假定这些作业在单机上需要很长的运行时间，因此使用多台机器可以缩短运行时间。

1．MapReduce简介

MapReduce是一种分布式计算模型，在处理海量数据上具有很明显的优势，因此常被用于大规模数据集的并行计算。MapReduce还是开源的，任何人都可以借助这个框架来进行并行编程，这个框架使得之前复杂的分布式编程变得相当容易实现。

MapReduce分布式编程模型是作为谷歌引以为豪的三大云计算相关的核心技术（GFS、Big Table和MapReduce）之一，被设计用于并行运算处理大于1TB的海量数据集。MapReduce的最初灵感来源于函数式编程语言中经常用到的映射（Map）和规约（Reduce）函数。它将复杂的并行算法处理过程抽象为一组概念简单的接口，用来实现大规模海量信息处理的并行化和分布化，从而使得没有多少并行编程经验的开发人员也能轻松地进行并行编程。

MapReduce分布式编程模型可以用于能灵活调整的普通计算机所构成的中小规模集群之上，典型的MapReduce系统能运行于由数以千计普通计算机所组成的集群中，这已经在谷歌中得到了实现与应用。MapReduce分布式编程模型的主要贡献在于：通过实现一组概念简单却又强大的接口以实现大规模计算的并行化和分布化，并通过实现这些接口，MapReduce能够组建由普通计算机作为成员的高性能集群。在采用MapReduce分布式模式的系统上，每一个单独的节点上均可以同时运行一个Map任务和一个Reduce任务，所以MapReduce的处理效率非常高。

2．MapReduce的发展历史

MapReduce出现的历史要追溯到1956年，图灵奖获得者、著名的人工智能专家McCarthy首次提出了LISP语言的构想，而在LISP语言中就包含了现在所使用的MapReduce功能。LISP语言是一种用于人工智能领域的语言，在人工智能领域有很多的应用。LISP在1956年设计时主要是希望有效地进行“符号运算”。它是一种表处理语言，其逻辑简单但结构不同于其他高级语言。1960年，McCarthy更是极有预见性地提出“今后计算机将会作为公共设施提供给公众”的观点。这一观点与现在人们对云计算的定义极为相近，所以McCarthy被称为“云计算之父”。MapReduce在McCarthy提出时并没有考虑到其在分布式系统和大数据上会有如此大的应用前景，只是作为一种函数操作来定义的。

2004年，谷歌公司的Dean发表文章将MapReduce这一编程模型在分布式系统中的应用进行了介绍，从此MapReduce分布式编程模型进入了人们的视野。可以认为分布式MapReduce是由谷歌公司首先提出的，Hadoop跟进了谷歌的这一思想。Hadoop是一个开源版本的谷歌系统，正是由于Hadoop的跟进普通用户才得以开发自己的基于MapReduce框架的云计算应用系统。

3．MapReduce的优缺点

MapReduce的优点主要有两个方面：MapReduce分布式处理框架，不仅能处理大规模数据，而且能将很多烦琐的细节隐藏起来，如自动并行化、负载均衡和灾备管理等，这样将极大地简化程序员的开发工作；MapReduce的伸缩性非常好，也就是说，每增加一台服务器，MapReduce能将差不多的计算能力接入集群中，而过去的大多数分布式处理框架在伸缩性方面都与MapReduce相差甚远。

但是MapReduce毕竟是一个离线计算框架，其不足之处主要有以下几个。

1）启动时间长。一个Map和Reduce作业前有启动任务环节，后有清理任务环节，这就使得最简单的作业也会消耗几秒钟的时间。

2）调度开销大。一个作业包含很多任务时，Hadoop将任务调度到各个节点上会消耗比较长的时间，当资源不足时作业还得排队。

3）短作业处理效率低。由于作业要容错，计算的中间结果要写回文件系统，这导致了不必要的输入/输出操作，严重降低短作业的处理速度。

4）数据必须先存储才能运算。MapReduce在搜索的应用中，先将爬虫得到的网页数据放在一个分布式存储系统中，然后间断性地对这些数据进行批量处理，即先存储数据，后对数据进行运算。

3.4.2 MapReduce设计方式

MapReduce是一个简单、方便的分布式编程模型，主要面向存储在HDFS中的数据。采用“分而治之”的思想，MapReduce将一个大规模数据分解为多个小规模数据，并将其分发给集群中的多个节点共同完成，这样可以有效降低每一部分的运算复杂度，达到提高运算效率的目的。

1．MapReduce的集群结构

一个MapReduce任务需要由以下四个部分协作完成。

1）客户端。客户端与集群进行交互的接口，可以进行任务提交、结果获取等工作。

2）JobTracker。JobTracker是集群的总负责节点，主要起到集群的调度作用，一个集群中只能有一个JobTracker。

3）TaskTracker。TaskTracker是作业的真正执行者，可以执行两类任务：Map任务或Reduce任务。执行Map任务的被称为Mapper，执行Reduce任务的被称为Reducer。一个集群中可以有多个TaskTracker。

4）分布式文件系统。分布式文件系统用来存储输入/输出的数据，通常使用HDFS。

2．MapReduce的执行过程

MapReduce的编程框架是由一个单独运行在主节点上的JobTracker和运行在每个集群从节点上的TaskTracker共同组成的。用户用map和reduce两个函数来表达计算。map函数的输入是一个<key,value>键值对，输出一个<key,value>键值对的集合的中间结果。MapReduce集合所有相同key值的value，然后提供给reduce函数。reduce函数收到key值和对应的value的集合，通过计算得到较小的value值的集合。

MapReduce任务分为两个阶段：Map阶段和Reduce阶段。JobTracker将一个大规模的任务根据数据量分解，Map阶段执行分解后的小任务并得到中间结果，Reduce阶段负责把这些中间结果汇总。具体执行过程如下。

1）数据预处理。在任务开始前，首先调用类库，将输入文件分为多个分片。

2）任务分配。JobTracker为集群中空闲的节点分配Map任务或Reduce任务。设集群中有M个Map任务和R个Reduce任务（Reduce任务数通常小于Map任务数）。

3）Map任务。Mapper读取自己所属的文件分片，将每一条输入数据转换为<key, value>键值对，使用map函数对每一个键值对进行处理，得到一个新的<key,value>键值对，并作为中间结果缓存在当前节点上。

4）缓存文件定位。Map任务得到的中间结果被周期性地写入Mapper所在的本地硬盘中，并把文件的存储位置信息经由JobTracker传递给Reducer。

5）Reducer拉取文件。Reducer通过位置信息到相应的Mapper处拉取这些文件，将同一key对应的所有取值合并，得到<key, list（value）>键值组。

6）Reduce任务。Reducer将所读取到的<key,list（value）>键值组使用reduce函数进行计算，得到最终结果并将其输出。

7）结束。当所有的Map任务和Reduce任务运行完毕后，系统会自动结束各个节点上的对应进程并将任务的执行情况反馈给用户。

每个Map操作都针对不同的初始数据，不同的Map操作之间彼此独立，互不影响，因而Map可以并行操作。Reduce操作是对Map操作之后产生的一部分结果进行规约操作。每个Reduce操作和Map操作一样也是互相独立的，所以Reduce也能够并行执行。由此得知，MapReduce编程框架的共同特征是MapReduce的数据均是被分割成设定大小的数据块，这些数据块均能够被并行处理。

MapReduce运行流程如图3-10所示。

MapReduce是一个简便的分布式编程框架，此框架下并行程序中需要map、reduce和main三个主要函数。编程人员只需实现其中的map函数和reduce函数，分布式存储、工作调度、负载平衡等其他问题均由MapReduce分布式框架负责完成。

3.4.3 MapReduce架构

在Hadoop的体系结构中，MapReduce是一个简单、易用的软件框架，基于MapReduce可以将任务分发到由上千台商用机器组成的集群上，并以一种可靠容错的方式并行处理大量的数据集，实现Hadoop的并行任务处理功能。

1．MapReduce架构

MapReduce主要采用Master/Slave（M/S）架构，其主要包括Client、JobTracker、TaskTracker和TaskScheduler四个组件。MapReduce架构如图3-11所示，下面分别对这四个组件进行介绍。

（1）Client

用户通过Client将编写的MapReduce程序提交到JobTracker端，作业运行状态也是通过Client提供的部分接口来查询的。在Hadoop内部，MapReduce程序是用作业（Job）表示的，一个MapReduce程序可对应若干个作业，而每个作业会被分解成若干个Map/Reduce任务（Task）。

（2）JobTracker

JobTracker主要实现资源监控和作业调度功能。JobTracker用来监控所有TaskTracker和作业的健康状况，在发现失败的情况下，JobTracker会跟踪其任务执行进度、资源使用量等信息，并将此信息生成报表发送给任务调度器，任务调度器在接收到命令之后及时选择合适的任务并将这些资源进行分配。在Hadoop中，任务调度器以模块形式存在，具有可插拔的特征，用户可以根据自己的需要设计相应的任务调度器。

（3）TaskTracker

TaskTracker使用Heartbeat将本节点上资源的使用情况和任务的进行情况汇报给JobTracker，同时接收JobTracker发送过来的命令并响应其启动新任务、杀死任务等操作。slot表示CPU、内存上的计算资源，slot可以帮助TaskTracker将每个节点上的资源进行等量划分，每个Task只有在获得slot才可以运行。slot包括Map slot和Reduce slot两种，Map slot供MapTask使用，Reduce slot供ReduceTask使用。

（4）TaskScheduler

Task可以分成MapTask和ReduceTask两种，且都由TaskTracker启动。HDFS是以块为固定大小来存储数据的，它是存储数据的基本单位。Split主要包括数据起始位置、数据长度和数据所在节点等基本的元数据信息，它是MapReduce的处理单元，每个Slip会由一个MapTask处理，Split的数量决定了MapTask的数目。MapTask将接收到对应的split通过迭代解析得到多个键值对，并使用用户自定义的map()函数来处理进程。经map()函数处理过的数据被分成多个partition，每个partition被对应的ReduceTask进行处理，并将数据保存在本地磁盘中。ReduceTask执行过程包括以下三个阶段。

1）从数据节点中读取MapTask的中间结果，此阶段称为“Shuffle阶段”。

2）根据键值对排序，此阶段称为“Sort阶段”。

3）依次读取key、value list的值，并调用用户自定义的函数reduce()处理结果，并将此结果保存到HDFS中，此阶段称为“Reduce阶段”。

2．MapReduce作业的生命周期

一个MapReduce作业的生命周期大体分为以下5个阶段。

（1）作业提交与初始化

用户在提交完作业之后，JobClient将程序JAR程序包、作业配置文件、分片元信息文件等作业相关信息上传至分布式文件系统上，分片元信息文件的作用是记录每个输入分片的逻辑位置信息。当JobTracker接收到JobClient的请求后，会立即进行初始化，之后在运行过程中需要监控作业运行情况，这就需要建立Job In Progress对象，而且可以同时监控多个任务的运行状况。

（2）任务调度与监控

JobTracker是用来对任务进行调度和监控的。TaskTracker通过Heartbeat周期性地向JobTracker发送本节点资源的使用情况，在有空闲资源的情况下，任务调度命令JobTracker按照一定的计划来选择合适的空闲资源。任务调度器是具有双层架构、比较独立的结构，可以完成对任务的选择，选择任务需要充分考虑数据的本地性。此外，JobTracker的作用保证任务运行可以成功，并可以跟踪作业的整个运行过程。如果TaskTracker或者Task运行失败，则重新进行任务运行时间的计算；如果运行进度落后，也会重新进行计算；如果其他运行结束，就重新启动一个相同的Task；最终选取计算最快的Task结果作为最终结果。

（3）任务运行环境准备

通过启动JVM将资源进行隔离，这就基本准备好了运行环境，都是通过TaskTracker来实现的。TaskTracker为每个Task启动一个独立的JVM，它为了防止Task滥用资源，采用了操作系统进程来实现隔离。

（4）任务执行

TaskTracker准备好任务的执行环境之后，就可以执行任务了。在执行过程中，每个任务都汇报给TaskTracker之后再给JobTracker。

（5）作业完成

如果其中的所有任务都执行完成，整个作业就完成了。