hadoop_MapReduce源码解析.docxVIP

综述Hadoop源代码分为三大模块：MapReduce、HDFS和Hadoop Common。其中MapReduce模块主要实现了MapReduce模型的相关功能；HDFS模块主要实现了HDFS的相关功能；而Hadoop Common主要实现了一些基础功能，比如说RPC、网络通信等。在用户使用Hadoop MapReduce模型进行并行计算时，用户只需要写好Map函数、Reduce函数，之后调用JobClient将Job提交即可。在JobTracker收到提交的Job之后，便会对Job进行一系列的配置，然后交给TaskTracker进行执行。执行完毕之后，JobTracker会通知JobClient任务完成，并将结果存入HDFS中如图所示，用户提交Job是通过JobClient类的submitJob（）函数实现的。在Hadoop源代码中，一个被提交了的Job由JobInProgress类的一个实例表示。该类封装了表示Job的各种信息，以及Job所需要执行的各种动作。在调用submitJob（）函数之后，JobTracker会将作业加入到一个队列中去，这个队列的名字叫做jobInitQueue。然后，在JobTracker中，有一个名为JobQueueTaskScheduler的对象，会不断轮询jobInitQueue队列，一旦发现有新的Job加入，便将其取出，然后将其初始化。在Hadoop代码中，一个Task由一个TaskInProgress类的实例表示。该类封装了描述Task所需的各种信息以及Task执行的各种动作。TaskTracker自从启动以后，会每隔一段时间向JobTracker发送消息，消息的名称为“Heartbeat”。Heartbeat中包含了该TaskTracker当前的状态以及对Task的请求。JobTracker在收到Heartbeat之后，会检查该heartbeat的里所包含的各种信息，如果发现错误会启动相应的错误处理程序。如果TaskTracker在Heartbeat中添加了对Task的请求，则JobTracker会添加相应的指令在对Heartbeat的回复中。在Hadoop源代码中，JobTracker对TaskTracker的指令称为action，JobTracker对TaskTracker所发送来的Heartbeat的回复消息称为HeartbeatResponse。在TaskTracker内部，有一个队列叫做TaskQueue。该中包含了所有新加入的Task。每当TaskTracker收到HeartbeatResponse后，会对其进行检查，如果其中包含了新的Task，便将其加入到TaskQueue中。在TaskTracker内部，有两个线程不断轮询TaskQueue，一个是MapLauncher，另一个是ReduceLauncher。如果发现有新加入的Map任务，MapLauncher便将其取出并且执行。如果是Reduce任务，ReduceLauncher便将其取出执行。不论是Map Task还是Reduce Task，当他们被取出之后，都要进行本地化。本地化的意思就是将所有需要的信息，比如需要运行的jar文件、配置文件、输入数据等等，一起拷贝到本地的文件系统。这样做的目的是为了方便任务在某台机器上独立执行。本地化之后，TaskTracker会为每一个task单独创建一个jvm，然后单独运行。等Task运行完之后，TaskTracker会通知JobTracker任务完成，以进行下一步的动作。等到所有的Task都完成之后，Job也就完成了，此时JobTracker会通知JobClient工作完成。代码详细分析下面从用户使用Hadoop进行MapReduce计算的过程为线索，详细介绍Task执行的细节，并对Hadoop MapReduce的主要代码进行分析。启动Hadoop集群Hadoop集群的启动是通过在Master上运行start-all.sh脚本进行的。运行该脚本之后，Hadoop会配置一系列的环境变量以及其他Hadoop运行所需要的参数，然后在本机运行JobTracker和NameNode。然后通过SSH登录到所有slave机器上，启动TaskTracker和DataNode。因为本文只介绍Hadoop MapReduce模块，所以NameNode和DataNode的相关知识不再介绍。JobTracker启动以及Job的初始化org.apache.hadoop.mapred.JobTracker类实现了Hadoop MapReduce模型的JobTracker的功能，主要负责任务的接受，初始化，调度以及对TaskTracker的监控。JobTracker单独作为一个JVM运行，main函数