深入剖析MapReduce架构及原理

MapReduce应用场景

MapReduce 定义

Hadoop 中的 MapReduce 是一个使用简单的软件框架，基于它写出来的应用程序能够运行在由上千个商用机器组成的大型集群上，并以一种可靠容错式并行处理TB级别的数据集。

MapReduce 来源

Hadoop MapReduce 源于 Google 在2004年12月份发表的 MapReduce 论文。 Hadoop MapReduce 其实就是 Google MapReduce 的一个克隆版本。

MapReduce 特点

MapReduce 为什么如此受欢迎？尤其现在互联网+时代，互联网+公司都在使用 MapReduce。MapReduce 之所以如此受欢迎，它主要有以下几个特点。

1、MapReduce 易于编程。它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的 PC 机器运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。 就是因为这个特点使得 MapReduce 编程变得非常流行。

2、良好的扩展性。当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

3、高容错性。MapReduce 设计的初衷就是使程序能够部署在廉价的 PC 机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上面上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由 Hadoop 内部完成的。

4、适合 PB 级以上海量数据的离线处理。这里加红字体离线处理，说明它适合离线处理而不适合在线处理。比如像毫秒级别的返回一个结果，MapReduce 很难做到。

MapReduce 虽然具有很多的优势，但是它也有不擅长的地方。这里的不擅长不代表它不能做，而是在有些场景下实现的效果差，并不适合 MapReduce 来处理，主要表现在以下几个方面。

1、实时计算。MapReduce 无法像 Mysql 一样，在毫秒或者秒级内返回结果。

2、流式计算。流式计算的输入数据时动态的，而 MapReduce 的输入数据集是静态的，不能动态变化。这是因为 MapReduce 自身的设计特点决定了数据源必须是静态的。

3、DAG（有向图）计算。多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce 并不是不能做，而是使用后，每个MapReduce 作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

MapReduce的编程模型

MapReduce 实例

为了分析 MapReduce 的编程模型，这里我们以 WordCount 为实例。就像 Java、C++等编程语言的入门程序 hello word 一样，WordCount 是 MapReduce 最简单的入门程序。下面我们就来逐步分析。

1、场景：假如有大量的文件，里面存储的都是单词。

类似应用场景：WordCount 虽然很简单，但它是很多重要应用的模型。

1) 搜索引擎中，统计最流行的 K 个搜索词。

2) 统计搜索词频率，帮助优化搜索词提示。

2、任务：我们该如何统计每个单词出现的次数？

3、将问题规范为：有一批文件（规模为 TB 级或者 PB 级），如何统计这些文件中所有单词出现的次数。

4、解决方案：首先，分别统计每个文件中单词出现的次数；然后，累加不同文件中同一个单词出现次数。

MapReduce 执行流程

通过上面的分析可知，它其实就是一个典型的 MapReduce 过程。下面我们通过示意图来分析 MapReduce 过程：

上图的流程大概分为以下几步。

第一步：假设一个文件有三行英文单词作为 MapReduce 的Input（输入），这里经过 Splitting 过程把文件分割为3块。分割后的3块数据就可以并行处理，每一块交给一个 map 线程处理。

第二步：每个 map 线程中，以每个单词为key，以1作为词频数value，然后输出。

第三步：每个 map 的输出要经过 shuffling（混洗），将相同的单词key放在一个桶里面，然后交给 reduce 处理。

第四步：reduce 接受到 shuffling 后的数据， 会将相同的单词进行合并，得到每个单词的词频数，最后将统计好的每个单词的词频数作为输出结果。

上述就是 MapReduce 的大致流程，前两步可以看做 map 阶段，后两步可以看做 reduce 阶段。下面我们来看看 MapReduce 大致实现。

1、Input：首先 MapReduce 输入的是一系列key/value对。key表示每行偏移量，value代表每行输入的单词。

2、用户提供了 map 函数和 reduce 函数的实现：

map(k,v) ——> list(k1,v1)

reduce(k1,list(v1)) ——>(k2,v2)

map 函数将每个单词转化为key/value对输出，这里key为每个单词，value为词频1。(k1,v1)是 map 输出的中间key/value结果对。reduce 将相同单词的所有词频进行合并，比如将单词k1，词频为list(v1)，合并为(k2,v2)。reduce 合并完之后，最终输出一系列(k2,v2)键值对。

下面我们来看一下 MapReduce 的伪代码。

map(key,value)://map 函数，key代表偏移量，value代表每行单词

for each word w in value: //循环每行数据，输出每个单词和词频的键值对(w,1)

emit(w,1)

reduce(key,values)://reduce 函数，key代表一个单词，value代表这个单词的所有词频数集合

result=0

for each count v in values: //循环词频集合，求出该单词的总词频数，然后输出(key,result)

result+=v 

emit(key,result)

讲到这里，我们可以对 MapReduce 做一个总结。MapReduce 将 作业的整个运行过程分为两个阶段：Map 阶段和Reduce 阶段。

1、Map 阶段

Map 阶段是由一定数量的 Map Task 组成。这些 Map Task 可以同时运行，每个 Map Task又是由以下三个部分组成。

1) 对输入数据格式进行解析的一个组件：InputFormat。因为不同的数据可能存储的数据格式不一样，这就需要有一个 InputFormat 组件来解析这些数据的存放格式。默认情况下，它提供了一个 TextInputFormat 来解释数据格式。TextInputFormat 就是我们前面提到的文本文件输入格式，它会将文件的每一行解释成(key,value)，key代表每行偏移量，value代表每行数据内容。 通常情况我们不需要自定义 InputFormat，因为 MapReduce 提供了很多种InputFormat的实现，我们根据不同的数据格式，选择不同的 InputFormat 来解释就可以了。这一点我们后面会讲到。

2)输入数据处理：Mapper。这个 Mapper 是必须要实现的，因为根据不同的业务对数据有不同的处理。

3)数据分组：Partitioner。Mapper 数据处理之后输出之前，输出key会经过 Partitioner 分组或者分桶选择不同的reduce。默认的情况下，Partitioner 会对 map 输出的key进行hash取模，比如有6个Reduce Task，它就是模（mod）6，如果key的hash值为0，就选择第0个 Reduce Task，如果key的hash值为1，就选择第一个 Reduce Task。这样不同的 map 对相同单词key，它的 hash 值取模是一样的，所以会交给同一个 reduce 来处理。

2、Reduce 阶段

Reduce 阶段由一定数量的 Reduce Task 组成。这些 Reduce Task 可以同时运行，每个 Reduce Task又是由以下四个部分组成。

1) 数据运程拷贝。Reduce Task 要运程拷贝每个 map 处理的结果，从每个 map 中读取一部分结果。每个 Reduce Task 拷贝哪些数据，是由上面 Partitioner 决定的。

2) 数据按照key排序。Reduce Task 读取完数据后，要按照key进行排序。按照key排序后，相同的key被分到一组，交给同一个 Reduce Task 处理。

3) 数据处理：Reducer。以WordCount为例，相同的单词key分到一组，交个同一个Reducer处理，这样就实现了对每个单词的词频统计。

4) 数据输出格式：OutputFormat。Reducer 统计的结果，将按照 OutputFormat 格式输出。默认情况下的输出格式为 TextOutputFormat，以WordCount为例，这里的key为单词，value为词频数。

InputFormat、Mapper、Partitioner、Reducer和OutputFormat 都是用户可以实现的。通常情况下，用户只需要实现 Mapper和Reducer，其他的使用默认实现就可以了。

MapReduce 内部逻辑

下面我们通过 MapReduce 的内部逻辑，来分析 MapReduce的数据处理过程。我们以WordCount为例，来看一下mapreduce 内部逻辑，如下图所示：

MapReduce包含四个组成部分，分别为Client、JobTracker、TaskTracker和Task，下面我们详细介绍这四个组成部分。

1）Client 客户端

每一个 Job 都会在用户端通过 Client 类将应用程序以及配置参数 Configuration 打包成 JAR 文件存储在 HDFS，并把路径提交到 JobTracker 的 master 服务，然后由 master 创建每一个 Task（即 MapTask 和 ReduceTask） 将它们分发到各个 TaskTracker 服务中去执行。

2）JobTracker

JobTracke负责资源监控和作业调度。JobTracker 监控所有TaskTracker 与job的健康状况，一旦发现失败，就将相应的任务转移到其他节点；同时，JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在Hadoop 中，任务调度器是一个可插拔的模块，用户可以根据自己的需要设计相应的调度器。

3）TaskTracker

TaskTracker 会周期性地通过Heartbeat 将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等）。TaskTracker 使用“slot”等量划分本节点上的资源量。“slot”代表计算资源（CPU、内存等）。一个Task 获取到一个slot 后才有机会运行，而Hadoop 调度器的作用就是将各个TaskTracker 上的空闲slot 分配给Task 使用。slot 分为Map slot 和Reduce slot 两种，分别供Map Task 和Reduce Task 使用。TaskTracker 通过slot 数目（可配置参数）限定Task 的并发度。

4）Task

Task 分为Map Task 和Reduce Task 两种，均由TaskTracker 启动。HDFS 以固定大小的block 为基本单位存储数据，而对于MapReduce 而言，其处理单位是split。split 是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。但需要注意的是，split 的多少决定了Map Task 的数目，因为每个split 只会交给一个Map Task 处理。Split 和 Block的关系如下图所示：

Map Task 执行过程如下图 所示：由该图可知，Map Task 先将对应的split 迭代解析成一个个key/value 对，依次调用用户 自定义的map() 函数进行处理，最终将临时结果存放到本地磁盘上, 其中临时数据被分成若干个partition，每个partition 将被一个Reduce Task 处理。

Reduce Task 执行过程下图所示。该过程分为三个阶段：

①从远程节点上读取Map Task 中间结果（称为“Shuffle 阶段”）；

②按照key 对key/value 对进行排序（称为“Sort 阶段”）；

③依次读取< key, value list>，调用用户自定义的reduce() 函数处理，并将最终结果存到HDFS 上（称为“Reduce 阶段”）。

MapReduce 数据本地性

在介绍数据本地性之前，我们首先介绍网络拓扑的概念。在一个 Hadoop 集群里面，通常我们把这些机器按照机架来组织，比如说每个机架一般有16-64个节点。每个机架通过 Switch 交换机来通信，不同的机架又通过总的Switch交换机来交互。其架构图如下所示：

在上图的架构中，我们标注了A、B、C、D四个节点，其中A、B、C三个节点存储了block1数据块。我们假设节点A要读取block1数据块，那么它的最佳选择就是读取它本身存储的block1，此时读取的速度最快、效率最高。如果A本身的block1数据块丢失或者损坏，它就会选择读取同一个Switch机架下的B节点上的block1，这是因为在同一个机架下面读取数据相对较快，所以不会选择跨机架读取C节点上面的block1，因为跨机架读取数据的效率最差。除非A、B中的block1都丢失或者损坏，才会选择跨机架读取C节点上面的block1。

数据本地性（data locality）

那到底什么是数据本地性呢？如果一个任务运行在它将处理的数据所在的节点，我们就称该任务具有“数据本地性”。数据的本地性可避免跨节点或者机架进行数据传输，提高运行效率。

数据本地性分为三个类别：

1、同节点（node-local）。

2、同机架（rack-local）。

3、跨机架（off-switch）。

为了深入理解数据本地性的三个类别，我们下面举个示例，如下图所示：