hadoop作业提交 初始化

下面我们来熟悉一下MapReduce作业的提交和初始化的过程： 
 

        作业的提交过程 
        第一步，使用Hadoop提供的shell命令提交作业，命令行作业提交命令：

$HADOOP_HOME/bin/hadoop jar job.jar \
    -D mapred.job.name="task-test" \
    -D mapred.reduce.tasks=3 \
    -files=blacklist.txt,whitelist.xml \
    -libjars=xxx.jar \
    -archives=yyy.zip \
    -input /test/input \
    -output /test/output

        当用户按上述命令格式提交作业后，命令行脚本会调用JobClient.runJob()方法(1.0新版本的MapReduce API使用job.waitForCompletion(true)方法)提交作业，之后的函数调用过程如下图所示： 
 
        第二步，作业文件上传 
        JobClient将作业提交到JobTracker节点上之前，需要作业写初始化工作。初始化工作由JobClient.submitJobInternal(job)实现，这些初始化包括获取作业的jobId、创建HDFS目录、上传作业以及生成所有的InputSplit分片的相关信息等。 
        MapReduce的作业文件的上传和下载都是由DistributedCache透明完成的，它是Hadoop专门开发的数据分发工具。 

        第三步，生成InputSplit文件 
        作业提交后，JobClient会调用InputFormat的getSplits()方法生成相关的split分片信息，该信息包括InputSplit元数据信息和原始的InputSplit信息，其中元数据信息被JobTracker使用，第二部分在Map Task初始化时由Mapper使用来获取自己要处理的数据，这两部分数据被保存到job.split文件和job.splitmetainfo文件中。 

        第四步，将作业提交到JobTracker 
        JobClient通过RPC将作业提交到JobTracker端，在这一阶段会依次进行如下操作： 
        1）、为作业创建JobInProgress对象。JobTracker会为用户提交的每一个作业创建一个JobInProgress对象，这个对象维护了作业的运行时信息，主要用于跟踪正在运行的作业的状态和进度； 
        2）、检查用户是否具有指定队列的作业提交权限。Hadoop以队列为单位来管理作业和资源，每个队列分配有一定量的资源，管理严可以为每个队列指定哪些用户有权限提交作业； 
        3）、检查作业配置的内存使用量是否合理。用户在提交作业时，可已分别通过参数mapred.job.map.memory.mb和mapred.job.reduce.memory.mb指定Map Task和Reduce Task的内存使用量，而管理员可以给集群中的Map Task和Reduce Task分别设置中的内存使用量，一旦用户配置的内存使用量超过总的内存限制，作业就会提交失败； 
        4）、通知TaskScheduler初始化作业 
        JobTracker收到提交的作业后，会交给TaskScheduler调度器，然后按照一定的策略对作业做初始化操作。 

        作业的初始化 
        作业的初始化主要是指构造Map Task和Reduce Task并对它们进行初始化操作。 
        这一步的操作主要是由调度器调用JobTracker.initJob()方法来对新作业做初始化的。Hadoop将每个作业分节成4中类型的任务：Setup Task，Map Task，Reduce Task和Cleanup Task，它们的运行时信息由TaskInProgress维护，因此，从某个方面讲，创建这些任务就是创建TaskInProgress对象。 
        Setup Task 
        作业初始化标志性任务，它进行一些很简单的作业初始化工作。该类型任务又分为Map Setup Task和Reduce Setup Task两种，并且只能运行一次。 
        Map Task 
        Map阶段的数据处理任务。 
        Reduce Task 
        Reduce阶段的处理数据的任务。其数目可以由用户通过参数mapred.reduce.tasks指定。Hadoop刚开始的时候只会调度Map Task任务，直到Map Task完成数目达到由参数mapred.reduce.slowstart.completed.maps指定的百分比后，才开始调度Reduce Task。 
        Cleanup Task 
        作业结束的标志性任务，主要是做一些作业清理的工作，比如删除作业在运行中产生的一些零食目录和数据等信息。

 

总结：

Job作业提交 
        1、 编写MapReduce job程序，通过命令行向Hadoop提交并开始执行作业：

引用

        # hadoop  fs  -put  xxx.jar  /path      //或者使用hadoop  fs  -copyFromLocal  xxx.jar  /path 向copy文件 
        # chown -R hadoop xxx.jar 
        # chgrp -R hadoop xxx.jar 
        # hadoop jar xxx.jar package.ClassName input_path output_path 
       

        客户端实际上是通过JobClient的submitJobInternal提交作业的。 
        2、本次运行，NameNode节点会首先检查用户权限和提交参数是否正确，通过后生成一个job，于是JobClient向JobTracker申请一个JobID用于标识这个job; 
        3、JobClient将Job所需要的资源提交到HDFS中一个以JobID命名的目录中，这些资源包括Jar包、配置文件、InputSplit元数据信息等； 
        4、JobClient通过调用submitJob()向JobTracker告知Job也准备好，提交这个Job； 

        Job初始化 
        5、JobTracker初始化Job，当JobTracker接收到客户端submitJob()方法，会将该作业提交到内部的作业调度队列，并做相关的初始化操作，如生成JobInProgress对象和Map或Reduce Task列表等； 
        6、JobTracker通过JobID从HDFS获取这个Job的Split等信息； 
        
        任务分配(Task Assignment) 
        7、TaskTrackers节点周期性的通过心跳Heartbeat向JobTracker节点汇报自己的相关信息，如“自己”还活着、可以接受新任务等，然后然后JobTracker或根据一定的作业调度算法给TaskTrackers节点分配一个Map Task和Reduce Task任务； 

        任务执行(Task Execution) 
        8、TaskTracker节点在获得JobTracker指派给的任务后，会到中获取当前任务相关的资源； 
        9、TaskTracker启动会给分配给自己的任务启动一个JVM； 
        10、最后在新启动的JVM中开始执行Mapper或Reducer Task任务。