全面讲解MapReduce处理流程

     今天，课课家和大家一起讨论“MapReduce处理流程”，这些都是一些容易理解的知识，这方面的大神还请绕道，当然也可以交流一下。有需要的小伙伴可以参考一下。

     MapReduce是Hadoop2.x的一个计算框架，利用分治的思想，将一个计算量很大的作业分给很多个任务，每个任务完成其中的一小部分，然后再将结果合并到一起。将任务分开处理的过程为map阶段，将每个小任务的结果合并到一起的过程为reduce阶段。下面先从宏观上介绍一下客户端提交一个作业时，Hadoop2.x各个组件之间的联系及处理流程。然后我们再具体看看MapReduce计算框架在执行一个作业时，做了些什么。

　　YARN

　　YARN是Hadoop2.x框架下的资源管理系统，其组成部分为：

　　1)全局资源管理器(global resource manager)：整个系统的资源管理和调配。

　　2)节点管理器(node manager)(每个节点都有一个)负责任务的启动、配置及其资源的监控

　　3)针对每个应用程序的应用程序管理器(application-specific application master)(因为Hadoop2.x支持的计算框架有很多，不只是MapReduce，还有像storm、spark、Tez不同处理机制的计算框架，所以MapReduce是一种应用程序，每个MapReduce作业是MapReduce类型程序的一个实例)

　　4)调度器(scheduler)(在资源管理器里)

　　5)容器(container)：一部分CPU和内存组成一个容器，最为资源使用，一个应用程序运行在一组容器中。

　　在了解了各个组件的功能之后，借助下图，我们看一下提交一个作业的流程：

　　1)客户端向资源管理器提交作业程序，作业程序的类型决定了使用哪种应用程序管理器(MapReduce、storm、Tez...)

　　2)资源管理器协调资源，在一个节点上获取一个运行应用程序管理器实例的容器

　　3)应用程序管理器(application master)在资源管理器中注册

　　4)应用程序管理器通过资源请求与资源管理器协商资源，包括该容器所在的节点和该容器的详细说明(CPU核数量和内存大小)

　　5)和 6)应用程序管理器在一个节点上或者多个节点上运行其Map Task和Reduce Task

　　7)在容器中运行的应用程序向应用程序管理器汇报执行度

　　8)应用程序执行完毕，应用程序管理器就会从资源管理器中取消注册，作业占用的资源会释放到系统中

　　MapReduce计算框架

　　MapReduce总的可以分为map阶段、shuffle阶段和reduce阶段。

　　map阶段

　　1)从HDFS中将输入值传输到Mapper节点

　　除了传输之外，在读取过程中，还需要做一个转换过程，将数据转换为键值对的形式(MapReduce处理的输入必须为键值对的形式)，这个过程通过InputFormat完成(默认为TextInputFormat)

　　2)Mapper

　　根据自己写的Mapper函数对文件进行处理，同样输出的是键值对(如wordcount中统计收到的数据中每个词出现的次数)

　　3)Partitioner

　　Patitioner根据Reducer的数量和自定义的划分方法(没有自定义的话，Hadoop有默认实现)去划分Mapper的输出;划分的结果会按照Mapper输出的键进行排序。

　　4)Combiner(这一步是可选的)

　　经过Partitioner排序后，如果作业中配置了Combiner，就会调用Combiner，Combiner就好像在Mapper端提前进行一下Reducer一样。

　　那为什么要提前进行呢?这是为了尽量减少对网络带宽的需求，比如经典的wordcount程序，在Mapper端处理之后，我们可能得到一个像key = apple，value = {1,1,1,1,1,1}的结果，如果我们能先对其进行一下Combiner，那么就能得到key = apple，value = 6的结果，传输这样的数据，肯定是要比key = apple，value = {1,1,1,1,1,1}的数据节省带宽的。

　　那既然能够节省传输带宽，为什么又是可选的呢?何不每次都默认执行Combiner?这是因为并不是每一个Mapper都能进行Combiner;比如现在我们的任务要统计一段时间内的每天的最高气温，假设开始有两个Mapper，输出为(0，10，20)和(15，25)，那么提前进行Combiner可以使得传递给Reducer端的数据为(20， 25)这样最后的结果还是为25，且传输的数据量变小;但是假如我们要求一段时间内的平均温度呢?如果开始就在Mapper端进行Combiner求平均温度，那么Reducer端得到的数据为(10, 20)，算出的平均温度为15，但是实际上的平均温度为(0， 10， 20 ， 15 ，25)的平均，为14;所以需要搞清楚Combiner合适不合适提前进行。

　　MapReduce主要功能

　　MapReduce提供了以下的主要功能：

　　1）数据划分和计算任务调度：

　　系统自动将一个作业（Job）待处理的大数据划分为很多个数据块，每个数据块对应于一个计算任务（Task），并自动调度计算节点来处理相应的数据块。作业和任务调度功能主要负责分配和调度计算节点（Map节点或Reduce节点），同时负责监控这些节点的执行状态，并负责Map节点执行的同步控制。

　　2）数据/代码互定位：

　　为了减少数据通信，一个基本原则是本地化数据处理，即一个计算节点尽可能处理其本地磁盘上所分布存储的数据，这实现了代码向数据的迁移；当无法进行这种本地化数据处理时，再寻找其他可用节点并将数据从网络上传送给该节点（数据向代码迁移），但将尽可能从数据所在的本地机架上寻找可用节点以减少通信延迟。

　　3）系统优化：

　　为了减少数据通信开销，中间结果数据进入Reduce节点前会进行一定的合并处理；一个Reduce节点所处理的数据可能会来自多个Map节点，为了避免Reduce计算阶段发生数据相关性，Map节点输出的中间结果需使用一定的策略进行适当的划分处理，保证相关性数据发送到同一个Reduce节点；此外，系统还进行一些计算性能优化处理，如对最慢的计算任务采用多备份执行、选最快完成者作为结果。

　　4）出错检测和恢复：

　　以低端商用服务器构成的大规模MapReduce计算集群中，节点硬件（主机、磁盘、内存等）出错和软件出错是常态，因此MapReduce需要能检测并隔离出错节点，并调度分配新的节点接管出错节点的计算任务。同时，系统还将维护数据存储的可靠性，用多备份冗余存储机制提高数据存储的可靠性，并能及时检测和恢复出错的数据。

　　shuffle阶段

　　shuffle阶段要做的事就是保证Mapper输出的数据传输到合适的Reducer进行处理，如下图所示：

　　shuffle阶段，每个Reducer都会使用HTTP协议从Mapper节点获得自己的划分(Reducer通过Application Master来获取自己应该查询哪些Mapper节点来获取自己划分的信息，因为每个Mapper实例完成后，会通知Application Master运行阶段产生的划分)

　　reduce阶段

　　1)Reducer

　　根据自己写的reduce程序对数据进行处理(如wordcount中将每个单词出现的次数加起来得到总和)

　　2)将处理结果输出到HDFS

　　通过OutputFormat完成(默认是TextOutputFormat)

　　总结

　　通过对Hadoop2.x框架的处理流程和MapReduce计算框架的处理流程的梳理，可以在进行程序编写时有一个更清楚的认识，下一步应该具体做些什么。

     结束语：看完文章的小伙伴，了解“MapReduce处理流程”了吧！有什么问题可以提出来和大家交流一下。如果各位小伙伴还想了解更多关于这方面的知识，随时可以登陆课课家，这里有全面的知识内容还有视频教学，随时欢迎你的到来哟~