Hadoop中的MapReduce是一个使用简单的软件框架，基于它写出来的应用程序能够运行在由上千个机器组成的大型集群上，并且以一种可靠容错并行处理TB级别的数据集。
一个MapReduce作业(job)通常会把输入的数据集切分为若干独立的数据块，由Map任务并行处理它们。框架会对map函数的输出先进行排序，然后把结果输入 给Reduce任务。通常作业的输入和输出都会被存储在文件系统中。整个框架负责任务的调度和和监控，以及重新执行已经失败的任务。
通常，MapReduce框架和分布式文件系统是运行在一组相同的节点上的，也就是说，计算节点和存储节点通常在一起。这种配置允许框架在那些已经存好数据的节点上高效地调度任务 ，这可以使整个集群的网络带宽被非常高效地利用。下面让我们一起看 MapReduce 是如何工作的？
1、MapReduce 的运行机制
下面我们主要从两个方面来探讨 MapReduce 的运行机制。
1、从客户端、JobTracker、TaskTracker的层次来分析MapReduce的工作原理，其原理图如下所示
从上图可以看出，MapReduce工作原理大致可以分为以下几个步骤。
1、首先在客户端启动一个作业。
2、然后向JobTracker请求一个Job ID。
3、接着将运行作业所需要的资源文件复制到HDFS上，包括MapReduce程序打包的JAR文件、配置文件和客户端计算所得的输入划分信息。这些文件都存放在JobTracker专门为该作业创建的文件夹中。文件夹名为该作业的Job ID。JAR文件默认会有10个副本（mapred.submit.replication属性控制）；输入划分信息告诉了JobTracker应该为这个作业启动多少个map任务等信息。
4、JobTracker接收到作业后，将其放在一个作业队列里，等待作业调度器对其进行调度，当作业调度器根据自己的调度算法调度到该作业时，会根据输入划分信息为每个划分创建一个map任务，并将map任务分配给TaskTracker执行。对于map和reduce任务，TaskTracker根据主机核的数量和内存的大小有固定数量的map槽和reduce槽。这里需要强调的是：map任务不是随随便便地分配给某个TaskTracker的， 这里有个概念叫：数据本地化 （Data-Local）。意思是：将map任务分配给含有该map处理的数据块的TaskTracker上，同时将程序JAR包复制到该TaskTracker上来运行，这叫"运算移动，数据不移动"。而分配reduce任务时并不考虑数据本地化。
5、TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带着很多的信息，比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成"成功"。当JobClient查询状态时，它将得知任务已完成，便显示一条消息给用户
2、从map任务和reduce任务的层次来分析MapReduce的运行机制，其原理图如下所示
从上图可以看出，MapReduce 工作原理大致可以分为两个阶段：Map阶段和Reduce阶段。
Map 阶段
1、每个输入分片会让一个map任务来处理，默认情况下，以HDFS的一个块的大小（默认为64M）为一个分片，当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中（该缓冲区的大小默认为100M，由io.sort.mb属性控制），当该缓冲区快要溢出时（默认为缓冲区大小的80%，由io.sort.spill.percent属性控制），会在本地文件系统中创建一个溢出文件，将该缓冲区中的数据写入这个文件。
2、在写入磁盘之前，线程首先根据reduce任务的数目将数据划分为相同数目的分区，也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据，而有些reduce任务却分到很少数据，甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序，如果此时设置了Combiner，将排序后的结果进行Combiner操作，这样做的目的是让尽可能少的数据写入到磁盘。
3、当map任务输出最后一个记录时，可能会有很多的溢出文件，这时需要将这些文件合并。 合并的过程中会不断地进行排序和combiner操作，目的有两个：
1)尽量减少每次写入磁盘的数据量。
2)尽量减少下一复制阶段网络传输的数据量。
最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量，这里可以将数据压缩，只要将mapred.compress.map.out设置为true就可以了。
4、将分区中的数据拷贝给相对应的reduce任务。有人可能会问：分区中的数据怎么知道它对应的reduce是哪个呢？其实map任务一直和其父TaskTracker保持联系，而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了。
到这里，map端就分析完了。那到底什么是Shuffle呢？Shuffle的中文意思是"洗牌"，如果我们这样看：一个map产生的数据，结果通过hash过程分区却分配给了不同的reduce任务，是不是一个对数据洗牌的过程呢？下面我们来看 Reduce 端。
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函数
2、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。
1、数据本地性（data locality）
那到底什么是数据本地性呢？如果一个任务运行在它将处理的数据所在的节点，我们就称该任务具有"数据本地性"。数据的本地性可避免跨节点或者机架进行数据传输，提高运行效率。
2、数据本地性分为三个类别：
1、同节点（node-local）。
2、同机架（rack-local）。
3、跨机架（off-switch）。
为了深入理解数据本地性的三个类别，我们下面举个示例，如下图所示
从上图可以看出一共有四个机架：R1、R2、R3和R4，每个机架上有三个节点，每个节点上面存储有不同的数据块。比如，R1机架下有三个节点：H1、H2和H3，H1节点上有b1和b8数据块，H2节点上有b1和b2数据块，H3节点上有b2和b3数据块。从上图可以看出，每个数据块有3个备份，按照 HDFS 的备份机制将这3个数据块存储到不同的节点上。下面我们来看一下数据本地性的三种类别。
1、我们假设Task1需要处理b1，H1节点正好有空闲资源得到Task1，H1节点存储有b1，这时Task1在本节点上直接读取b1，效率最高。这种情况下数据本地性称之为node-local。
2、我们假设Task2需要处理b2，H4节点正好有空闲资源得到Task2，b2不在H4节点上，那么Task2会找到存储有b2且距离最近的H5节点，H4和H5处于同一个机架，这时Task2会读取同机架节点上的b2。这种情况下数据本地性称之为rack-local。
3、我们假设Task3需要处理b3，H7节点正好有空闲资源得到Task3，b3并不在H7节点上，也不在H7同机架的节点上，而是在R1和R2机架下面的节点上，这时Task3就需要跨机架读取b3，效率非常低下。这种情况下数据本地性称之为off-switch。
原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/7692.html