（第4篇）hadoop之魂--mapreduce计算框架，让收集的数据产生价值

摘要: 通过前面的学习，大家已经了解了HDFS文件系统。有了数据，下一步就要分析计算这些数据，产生价值。接下来我们介绍Mapreduce计算框架，学习数据是怎样被利用的。

博主福利 给大家赠送一套hadoop视频课程

授课老师是百度 hadoop 核心架构师
内容包括hadoop入门、hadoop生态架构以及大型hadoop商业实战案例。
讲的很细致， MapReduce 就讲了 15 个小时。
学完后可以胜任 hadoop 的开发工作，很多人学的这个课程找到的工作。
（包括指导书、练习代码、和用到的软件都打包了）
先到先得先学习。联系老师微信ganshiyu1026，备注OSchina。即可免费领取

部分视频截图展示





通过前面的学习，大家已经了解了HDFS文件系统。有了数据，下一步就要分析计算这些数据，产生价值。接下来我们介绍Mapreduce计算框架，学习数据是怎样被利用的。

Mapreduce计算框架

如果将Hadoop比做一头大象，那么MapReduce就是那头大象的电脑。MapReduce是Hadoop核心编程模型。在Hadoop中，数据处理核心就是MapReduce程序设计模型。

本章内容：

1) MapReduce编程模型

2) MapReduce执行流程

3) MapReduce数据本地化

4) MapReduce工作原理

5) MapReduce错误处理机制

1. MapReduce编程模型

Map和Reduce的概念是从函数式变成语言中借来的，整个MapReduce计算过程分为Map阶段和Reduce阶段，也称为映射和缩减阶段，这两个独立的阶段实际上是两个独立的过程，即Map过程和Reduce过程，在Map中进行数据的读取和预处理，之后将预处理的结果发送到Reduce中进行合并。

我们通过一个代码案例，让大家快速熟悉如何通过代码，快速实现一个我们自己的MapReduce。

案例：分布式计算出一篇文章中的各个单词出现的次数，也就是WordCount。

1) 创建map.py文件，写入以下代码：

#!/usr/bin/env python import sys word_list = [] for line in sys.stdin: word_list = line.strip().split(' ') if len(word_list) <= 0: continue for word in word_list: w = word.strip() if len(w) <= 0: continue print '\t'.join([w, "1"])

该代码主要工作是从文章数据源逐行读取，文章中的单词之间以空格分割，

word_list = line.strip().split(' ')这块代码是将当前读取的一整行数据按照空格分割，将分割后的结果存入word_list数组中，然后通过for word in word_list遍历数组，取出每个单词，后面追加“1”标识当前word出现1次。

2) 创建reduce.py，写入以下代码：

#!/usr/bin/env python import sys cur_word = None sum_of_word = 0 for line in sys.stdin: ss = line.strip().split('\t') if len(ss) != 2: continue word = ss[0].strip() count = ss[1].strip() if cur_word == None: cur_word = word if cur_word != word: print '\t'.join([cur_word, str(sum_of_word)]) sum_of_word = 0 cur_word = word sum_of_word += int(count) print '\t'.join([cur_word, str(sum_of_word)]) sum_of_word = 0

该代码针对map阶段的数组进行汇总处理，map到reduce过程中默认存在shuffle partition分组机制，保证同一个word的记录，会连续传输到reduce中，所以在reduce阶段只需要对连续相同的word后面的技术进行累加求和即可。

3) 本地模拟测试脚本：

]$ cat big.txt | python map.py | sort -k1 | python reduce.py cat 1 run 3 see 2 spot 2 the 1

6) 脚本执行流程：

2. MapReduce执行流程

上面的例子属于MapReduce计算框架的一般流程，经过整理总结：





1) 输入和拆分：

不属于map和reduce的主要过程，但属于整个计算框架消耗时间的一部分，该部分会为正式的map准备数据。

分片（split）操作：

split只是将源文件的内容分片形成一系列的InputSplit，每个InputSpilt中存储着对应分片的数据信息（例如，文件块信息、起始位置、数据长度、所在节点列表…），并不是将源文件分割成多个小文件，每个InputSplit都由一个mapper进行后续处理。

每个分片大小参数是很重要的，splitSize是组成分片规则很重要的一个参数，该参数由三个值来确定：

l minSize：splitSize的最小值，由mapred-site.xml配置文件中mapred.min.split.size参数确定。

l maxSize：splitSize的最大值，由mapred-site.xml配置文件中mapreduce.jobtracker.split.metainfo.maxsize参数确定。

l blockSize：HDFS中文件存储的快大小，由hdfs-site.xml配置文件中dfs.block.size参数确定。

splitSize的确定规则：splitSize=max{minSize，min{maxSize，blockSize}}

数据格式化（Format）操作：

将划分好的InputSplit格式化成键值对形式的数据。其中key为偏移量，value是每一行的内容。

值得注意的是，在map任务执行过程中，会不停的执行数据格式化操作，每生成一个键值对就会将其传入map，进行处理。所以map和数据格式化操作并不存在前后时间差，而是同时进行的。





2) Map映射：

是Hadoop并行性质发挥的地方。根据用户指定的map过程，MapReduce尝试在数据所在机器上执行该map程序。在HDFS中，文件数据是被复制多份的，所以计算将会选择拥有此数据的最空闲的节点。

在这一部分，map内部具体实现过程，可以由用户自定义。

3) Shuffle派发：

Shuffle过程是指Mapper产生的直接输出结果，经过一系列的处理，成为最终的Reducer直接输入数据为止的整个过程。这是mapreduce的核心过程。该过程可以分为两个阶段：

Mapper端的Shuffle：由Mapper产生的结果并不会直接写入到磁盘中，而是先存储在内存中，当内存中的数据量达到设定的阀值时，一次性写入到本地磁盘中。并同时进行sort（排序）、combine（合并）、partition（分片）等操作。其中，sort是把Mapper产生的结果按照key值进行排序；combine是把key值相同的记录进行合并；partition是把数据均衡的分配给Reducer。

Reducer端的Shuffle：由于Mapper和Reducer往往不在同一个节点上运行，所以Reducer需要从多个节点上下载Mapper的结果数据，并对这些数据进行处理，然后才能被Reducer处理。

4) Reduce缩减：

Reducer接收形式的数据流，形成形式的输出，具体的过程可以由用户自定义，最终结果直接写入hdfs。每个reduce进程会对应一个输出文件，名称以part-开头。

3. MapReduce数据本地化（Data-Local）

首先，HDFS和MapReduce是Hadoop的核心设计。对于HDFS，是存储基础，在数据层面上提供了海量数据存储的支持。而MapReduce，是在数据的上一层，通过编写MapReduce程序对海量数据进行计算处理。

在前面HDFS章节中，知道了NameNode是文件系统的名字节点进程，DataNode是文件系统的数据节点进程。

MapReduce计算框架中负责计算任务调度的JobTracker对应HDFS的NameNode的角色，只不过一个负责计算任务调度，一个负责存储任务调度。

MapReduce计算框架中负责真正计算任务的TaskTracker对应到HDFS的DataNode的角色，一个负责计算，一个负责管理存储数据。

考虑到“本地化原则”，一般地，将NameNode和JobTracker部署到同一台机器上，各个DataNode和TaskNode也同样部署到同一台机器上。





这样做的目的是将map任务分配给含有该map处理的数据块的TaskTracker上，同时将程序JAR包复制到该TaskTracker上来运行，这叫“运算移动，数据不移动”。而分配reduce任务时并不考虑数据本地化。

4. MapReduce工作原理

我们通过Client、JobTrask和TaskTracker的角度来分析MapReduce的工作原理：





首先在客户端（Client）启动一个作业（Job），向JobTracker请求一个Job ID。将运行作业所需要的资源文件复制到HDFS上，包括MapReduce程序打包的JAR文件、配置文件和客户端计算所得的输入划分信息。这些文件都存放在JobTracker专门为该作业创建的文件夹中，文件夹名为该作业的Job ID。JAR文件默认会有10个副本（mapred.submit.replication属性控制）；输入划分信息告诉了JobTracker应该为这个作业启动多少个map任务等信息。

JobTracker接收到作业后，将其放在一个作业队列里，等待作业调度器对其进行调度当作业调度器根据自己的调度算法调度到该作业时，会根据输入划分信息为每个划分创建一个map任务，并将map任务分配给TaskTracker执行。对于map和reduce任务，TaskTracker根据主机核的数量和内存的大小有固定数量的map槽和reduce槽。这里需要强调的是：map任务不是随随便便地分配给某个TaskTracker的，这里就涉及到上面提到的数据本地化（Data-Local）。

TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带着很多的信息，比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成“成功”。当JobClient查询状态时，它将得知任务已完成，便显示一条消息给用户。





如果具体从map端和reduce端分析，可以参考上面的图片，具体如下：

Map端流程：

1) 每个输入分片会让一个map任务来处理，map输出的结果会暂且放在一个环形内存缓冲区中（该缓冲区的大小默认为100M，由io.sort.mb属性控制），当该缓冲区快要溢出时（默认为缓冲区大小的80%，由io.sort.spill.percent属性控制），会在本地文件系统中创建一个溢出文件，将该缓冲区中的数据写入这个文件。

2) 在写入磁盘之前，线程首先根据reduce任务的数目将数据划分为相同数目的分区，也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据，而有些reduce任务却分到很少数据，甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序，如果此时设置了Combiner，将排序后的结果进行Combine操作，这样做的目的是让尽可能少的数据写入到磁盘。

3) 当map任务输出最后一个记录时，可能会有很多的溢出文件，这时需要将这些文件合并。合并的过程中会不断地进行排序和Combine操作，目的有两个：

l 尽量减少每次写入磁盘的数据量；

l 尽量减少下一复制阶段网络传输的数据量。

最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量，这里可以将数据压缩，只要将mapred.compress.map.out设置为true就可以了。

4) 将分区中的数据拷贝给相对应的reduce任务。分区中的数据怎么知道它对应的reduce是哪个呢？其实map任务一直和其父TaskTracker保持联系，而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就可以了。

Reduce端流程：

1) Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小，则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间的百分比），如果数据量超过了该缓冲区大小的一定比例（由mapred.job.shuffle.merge.percent决定），则对数据合并后溢写到磁盘中。

2) 随着溢写文件的增多，后台线程会将它们合并成一个更大的有序的文件，这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端，MapReduce都是反复地执行排序，合并操作，所以排序是hadoop的灵魂。

3) 合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce函数。

在Map处理数据后，到Reduce得到数据之前，这个流程在MapReduce中可以看做是一个Shuffle的过程。

在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对。到底当前的key应该交由哪个reduce去做呢，是需要现在决定的。 MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key 做hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。

5. MapReduce错误处理机制

MapReduce任务执行过程中出现的故障可以分为两大类：硬件故障和任务执行失败引发的故障。

1) 硬件故障

在Hadoop Cluster中，只有一个JobTracker，因此，JobTracker本身是存在单点故障的。如何解决JobTracker的单点问题呢？我们可以采用主备部署方式，启动JobTracker主节点的同时，启动一个或多个JobTracker备用节点。当JobTracker主节点出现问题时，通过某种选举算法，从备用的JobTracker节点中重新选出一个主节点。

机器故障除了JobTracker错误就是TaskTracker错误。TaskTracker故障相对较为常见，MapReduce通常是通过重新执行任务来解决该故障。

在Hadoop集群中，正常情况下，TaskTracker会不断的与JobTracker通过心跳机制进行通信。如果某TaskTracker出现故障或者运行缓慢，它会停止或者很少向JobTracker发送心跳。如果一个TaskTracker在一定时间内（默认是1分钟）没有与JobTracker通信，那么JobTracker会将此TaskTracker从等待任务调度的TaskTracker集合中移除。同时JobTracker会要求此TaskTracker上的任务立刻返回。如果此TaskTracker任务仍然在mapping阶段的Map任务，那么JobTracker会要求其他的TaskTracker重新执行所有原本由故障TaskTracker执行的Map任务。如果任务是在Reduce阶段的Reduce任务，那么JobTracker会要求其他TaskTracker重新执行故障TaskTracker未完成的Reduce任务。比如:一个TaskTracker已经完成被分配的三个Reduce任务中的两个，因为Reduce任务一旦完成就会将数据写到HDFS上，所以只有第三个未完成的Reduce需要重新执行。但是对于Map任务来说，即使TaskTracker完成了部分Map，Reduce仍可能无法获取此节点上所有Map的所有输出。所以无论Map任务完成与否，故障TaskTracker上的Map任务都必须重新执行。

2) 任务执行失败引发的故障

在实际任务中，MapReduce作业还会遇到用户代码缺陷或进程崩溃引起的任务失败等情况。用户代码缺陷会导致它在执行过程中抛出异常。此时,任务JVM进程会自动退出，并向TaskTracker父进程发送错误消息，同时错误消息也会写入log文件，最后TaskTracker将此次任务尝试标记失败。对于进程崩溃引起的任务失败，TaskTracker的监听程序会发现进程退出，此时TaskTracker也会将此次任务尝试标记为失败。对于死循环程序或执行时间太长的程序，由于TaskTracker没有接收到进度更新，它也会将此次任务尝试标记为失败，并杀死程序对应的进程。

在以上情况中，TaskTracker将任务尝试标记为失败之后会将TaskTracker自身的任务计数器减1，以便想JobTracker申请新的任务。TaskTracker也会通过心跳机制告诉JobTracker本地的一个任务尝试失败。JobTracker接到任务失败的通知后，通过重置任务状态，将其加入到调度队列来重新分配该任务执行（JobTracker会尝试避免将失败的任务再次分配给运行失败的TaskTracker）。如果此任务尝试了4次（次数可以进行设置）仍没有完成，就不会再被重试，此时整个作业也就失败了。





如果你认真学习了前面的文章，这时你已经了解了数据的存储，和数据的计算，已经对MapReduce计算框架，有了一个清晰的认识。

接下来我们就要开始学习应用于分布式应用的协作服务Zookeeper，它在hadoop中有起到了什么重要作用呢？