MapReduce任务执行总流程
一个MapReduce作业的执行流程是：代码编写 -> 作业配置 -> 作业提交 -> Map任务的分配和执行 -> 处理中间结果 -> Reduce任务的分配和执行 -> 作业完成，而在每个任务的执行过程中又包含输入准备 -> 任务执行 -> 输出结果。下图给出了MapReduce作业详细的执行流程图。
MapReduce作业执行流程图
1. 提交作业
一个MapReduce作业在提交到Hadoop之后会进入完全地自动化执行过程，在这个过程中，用户除了监控程序的执行情况和强制中止作业之外，不能对作业的执行流程进行任何的干预。所以在作业提交之前，用户需要配置一下内容：
（1）程序代码：这里指Map和Reduce函数的具体代码，是MapReduce作业对应的程序必不可少的部分，并且这部分代码的逻辑正确与否与运行结果直接相关。
（2）Map和Reduce接口配置：在MapReduce中，Map接口需要派生自Mapper<k1,v1,k2,v2>接口，而Reduce接口则需要派生自Reduce<k1,v1,k2,v2>接口。在调用这两个方法的时候需要配置它们的四个参数，分别是输入key的数据类型、输入value的数据类型、输出key-value对的数据类型和context实例，其中输入输出的数据类型要与继承时所设置的数据类型相同。还有一个要求就是Map接口的输出key-value类型和Reduce接口的输入key-value类型要对应，因为Map输出组合value之后，它们会变为Reduce的输入内容。
（3）输入输出路径：作业提交之前，还需要在主函数中设置MapReduce作业在Hadoop集群上的输入路径和输出路径（必须保证输出路径不存在，否则会报错）。具体代码如下：
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.addOutputPath(job, new Path(args[0]));
（4）其他类型设置，比如调用runJob方法：先要在主函数中配置 如Outout的key和value类型、作业名称、InputFormat和OutputFormat等，最后再调用JobClient和runJob方法。
配置完作业的所有内容并确认无误之后就可以运行作业了，即执行图中的步骤①。提交作业的流程如下：
（1）通过调用JobTracker对象的getNewJobId()方法从JobTracker处获取当前作业的ID号（图中步骤②）。
（2）检查作业相关路径。在代码中获取各个路径信息时会对作业的对于路径进行检查。比如，检查输出路径是否已经存在，若存在则作业不会被提交，并且会给MApReduce程序返回错误信息；再比如输入穆勒不存在或没有对于令牌也会返回错误等。
（3）计算作业的输入划分，并将划分信息写入Job.split文件，如果写入失败就会返回错误。split文件的信息主要包括:split文件头、split文件版本号、split的个数。这些信息中的每一条都会包括以下内容：split类型名（默认为FileSplit）、split的大小、split的内容（对于FileSplit来说是写入的文件名，此split在文件中的起始位置上）、split的location信息（即在哪个DataNode上）。
（4）将运行作业所需要的资源复制到对应的HDFS上：作业的JAR文件、配置文件、和计算所得到的输入划分等（图中步骤③）。
（5）调用JobTracker对象的submitJob()方法来真正提交作业，告诉JobTracker作业准备执行（图中步骤④）。
2. 初始化作业
在客户端用户作业调用JobTracker对象的submitJob()方法之后，JobTracker会把此调用放入内部的TaskTracker变量中，然后进行调度，默认的调度方法是JobQueueTaskScheduler，也就是FIFO调度方法。当客户作业被调度执行时，JobTracker会创建一个代表这个作业的JobInProgress对象，并将任务和记录信息封存到这个对象中，以便跟踪任务的状态和进程。季节下了JobInProgress对象的initTask函数会对任务进行初始化操作（图中步骤⑤）。详细步骤如下：
（1）从HDFS中读取作业对应的job.split（图中步骤⑥）。JobTracker会从HDFS中作业对应的路径获取JobClient在步骤③中写入的job.split文件，得到输入数据的划分信息，为后面的初始化过程中Map任务的分配做好准备。
（2）创建并初始化Map任务和Reduce任务。initTasks先根据输入数据划分信息中的个数设定Map Task的个数，然后为每个Map Task生成一个TaskInProgress来处理input split，并将Map Task放入nonRunningMapCache，以便在JobTracker向TaskTracker分配Map Task的时候使用。接下来根据JobConf中的mapred.reduce.tasks属性利用setNumReduceTasks()方法来设置reduce task的个数，然后采用类似Map Task的方法将Reduce Task放入nonRunningReduces中，以便向TaskTracker分配Reduce Task时使用。
（3）最后就是创建两个初始化Task，根据个数和输入划分已经配置的信息，并分别初始化Map和Reduce。
3. 分配任务
TaskTracker和JobTracker之间的通信和任务分配是通过心跳机制按成的。TaskTracker作为一个单独的JVM执行一个简单的循环，主要实现每隔一段时间向JobTracker发送心跳（Heartbeat），以次告诉JobTracker此TaskTracker是否存活，是否准备执行新的任务。JobTracker接收到心跳信息后，如果有待分配的任务，它就会为TaskTracker分配一个任务，并将分配信息封装在心跳通信的返回值中返回给TaskTracker。TaskTracker从心跳方法的Response中得知此TaskTracker需要做的事情，如果是一个新的Task，则将它加入本机的任务队列中（图中步骤⑦）。
4. 执行任务
TaskTracker申请到新的任务之后就要在本地运行任务。运行任务的第一步是将任务本地化：将任务运行所必需的数据、配置信息、程序代码从HDFS复制到TaskTracker本地（图中步骤⑧）。这个主要是通过调用localizeJob()方法来完成的。这个方法主要通过下面结果步骤来完成任务的本地化：
（1）将job.split复制到本地；
（2）将job.jar复制到本地；
（3）将job的配置信息写入job.xml；
（4）创建本地任务目录，解压job.jar；
（5）调用launchTaskForJob()方法发布任务（图中步骤⑨）。
任务 本地化之后就可以通过调用launchTaskForJob()真正启动起来。接下来launchTaskForJob()又会调用launchTask()方法启动任务。launchTask()方法会先为任务创建本地目录，然后启动TaskRunner。在启动TaskRunner后，对于Map任务会启动MapTaskRunner；对于Reduce任务则启动ReduceTaskRunner。之后，TaskRunner会启动新的JVM来运行每个任务（图中步骤⑩）。以Map任务为例，任务执行的简单流程如下：
（1）配置执行参数（获取Java程序的执行环境和配置参数等）；
（2）在Child临时文件表中添加Map任务信息（运行Map和Reduce任务的主要进程是Child类）；
（3）配置log文件夹，然后配置Map任务的通信和输出参数；
（4）读取input split，生成RecordReader读取数据；
（5）为Map任务生成MapRunnable，依次从RecordReader中接收数据，并调用Mapper的Map函数进行处理；
（6）最后将Map函数的输出调用collect收集到MapOutputBuffer中。
5. 更新任务执行进度和状态
由MapReduce作业分割成的每个任务中都有一组计数器，它们对任务执行过程中的进度组成时间进行计数。如果人妖要报告进度，它便会设置一个标志以表明状态变化，并将其发送到TaskTracker上。另一个监听线程检查到这个标志后会告知TaskTracker当前的任务状态。同时，TaskTracker在每隔5秒发送给JobTracker的心跳中分装任务状态，报告自己的任务执行状态。
通过心跳机制，所有的TaskTracker的统计信息都会汇总到JobTracker处。JobTracker将这些信息合并起来，产生一个全局作业进度统计信息，用来表明正在运行的所有作业，以及其中所含任务的状态。最后，JobClient通过每秒查看JobTracker来接收作业进度的最新状态。
6. 完成作业
所有TaskTracker任务的执行进度信息都会汇总到JobTracker处，当JobTracker接收到最后一个任务的已完成通知后，便把作业的状态设置为“成功”。然后，JobClient也将及时得到任务已经完成，它会显示一条信息告知用户作业完成，最后从runJob()方法处返回。
以上就是博主为大家介绍的这一板块的主要内容，这都是博主自己的学习过程，希望能给大家带来一定的指导作用，有用的还望大家点个支持，如果对你没用也望包涵，有错误烦请指出。如有期待可关注博主以第一时间获取更新哦，谢谢！
版权声明：本文为博主原创文章，未经博主允许不得转载。