MapReduce源码流程分析

《MapReduce源码流程分析》由会员上传分享，免费在线阅读，更多相关内容在教育资源-天天文库。

1、MapReduce源码流程分析MapReduce源码流程分析11.角色21.1JobClient21.2JobTracker21.3TaskTracker22.数据结构22.1JobInProgress22.2TaskInProgress22.3TaskLauncher32.4TaskTracker$TaskInProgress32.5Task32.6TaskRunner32.7JvmManager33.流程33.1Client端33.1.1初始化33.1.2与JobTracker端的交互43.2JobTra

2、cker端73.2.1启动初始化73.2.2与Client端交互103.2.3与TaskTracker端交互163.3TaskTracker端173.3.1启动初始化173.3.2与JobTracker端交互273.3.3运行任务2749/491.角色1.1JobClient每一个job都会在用户端通过JobClient类将应用程序以及配置参数打包成jar文件存储在HDFS，并把路径提交到JobTracker，然后由JobTracker创建每一个Task（即MapTask和ReduceTask）并将它们分发到

3、各个TaskTracker服务中去执行。1.2JobTrackerJobTracker是一个master（中心节点）服务，JobTracker负责生成和调度job的每一个子任务task运行于TaskTracker上，并监控它们，如果发现有失败的task就重新运行它。一般情况应该把JobTracker部署在单独的机器上。1.3TaskTrackerTaskTracker是运行于多个节点上的slaver（）服务。TaskTracker则负责直接执行每一个task。TaskTracker都需要运行在HDFS的Dat

4、aNode上。2.数据结构2.1JobInProgressJobClient提交job后，JobTracker会创建一个JobInProgress来跟踪和调度这个job，并把它添加到job队列里。JobInProgress会根据提交的jobjar中定义的输入数据集（已分解成FileSplit）创建对应的一批TaskInProgress用于监控和调度MapTask，同时在创建指定数目的TaskInProgress用于监控和调度ReduceTask，缺省为1个ReduceTask。2.2TaskInProgres

5、sJobTracker启动任务时通过每一个TaskInProgress来launchTask，这时会把Task对象（即MapTask和ReduceTask）序列化写入相应的TaskTracker服务中，TaskTracker收到后会创建对应的TaskInProgress（此TaskInProgress实现非JobTracker中使用的TaskInProgress，作用类似）49/49用于监控和调度该Task。启动具体的Task进程是通过TaskInProgress管理的TaskRunner对象来运行的。Tas

6、kRunner会自动装载jobjar，并设置好环境变量后启动一个独立的javachild进程来执行Task，即MapTask或者ReduceTask，但它们不一定运行在同一个TaskTracker中。2.3TaskLauncher2.4TaskTracker$TaskInProgress2.5Task2.6TaskRunner2.7JvmManager3.流程启动。3.1Client端3.1.1初始化客户端初始化主要是创建RPC远程服务器，在创建JobClient的时候进行创建的。是在客户端提交作业代码执行时

7、进行的，提交之后立刻客户端做必要的初始化之后再提交作业。JobClientjc=newJobClient(job);RunningJobrj=jc.submitJob(job);49/493.1.2与JobTracker端的交互客户端主要负责像JobTracker提交作业，在旧版的api中通过使用JonConf类对作业进行配置，JobConf的功能已被新的类Configuration和Job替换。Configuration类描述了资源，这些资源大多都是从XML配置文件中读取的属性和值组成。比如来自core-d

8、efault.xml和core-site.xml。Job描述了用户角度的视图，它允许用户配置、提交、控制它的执行和查询状态。如图所示：是一个MapReduce的程序客户端提交作业的代码，通过生成一个JobConf对作业进行配置，也可以用新的api生成一个Job（注释中），最后通过JobClient.runJob(conf)对作业进行提交（也可用新的apijob.waitForCompletion(t