JobClient学习------作业提交与初始化

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//conf就是作业的配置对象，读取core-site、core-default、hdfs-site/default、mapred-site/default文件里的配置信息        
 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
//args[]就是使用hadoop jar命令运行作业时输入/输出路径参数，这两个参数传给了main函数
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);//System.exit(0)表示正常退出，exit()参数非0表示非正常退出。
    }

    Job job = new Job(conf, "word count");    
//以下就是设置job的一些运行参数，这些方法内部都会调用jobconf对象对应的方法
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
//以下就是设置job的一些运行参数，这些方法内部都会调用jobconf对象对应的方法
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

   可以看出作业完成了配置之后，调用了System.exit(job.waitForCompletion(true) ? 0 : 1);job.waitForCompletion(true)，true表示显示作业运行过程中的信息，waitForCompletion方法实现如下：

public boolean waitForCompletion(boolean verbose
                                   ) throws IOException, InterruptedException,
                                            ClassNotFoundException {
    if (state == JobState.DEFINE) {
      submit();
    }
    if (verbose) {
      jobClient.monitorAndPrintJob(conf, info);
    } else {
      info.waitForCompletion();
    }
    return isSuccessful();
  }

提交作业的时候，其状态是JobState.DEFINE，然后调用submit()方法，其实现如下所示：

public void submit() throws IOException, InterruptedException, 
                              ClassNotFoundException {
    ensureState(JobState.DEFINE);
    setUseNewAPI();//使用新API
    
    // Connect to the JobTracker and submit the job
    connect();
    info = jobClient.submitJobInternal(conf);
    super.setJobID(info.getID());
    state = JobState.RUNNING;
   }

使用connect()方法连接JobTracker，其实现如下：

 private void connect() throws IOException, InterruptedException {
    ugi.doAs(new PrivilegedExceptionAction<Object>() {
      public Object run() throws IOException {
        jobClient = new JobClient((JobConf) getConfiguration());    
        return null;
      }
    });
  }

public JobClient(JobConf conf) throws IOException {
    ......
    init(conf);
  }
public void init(JobConf conf) throws IOException {
     ......
     this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
  }
  private static JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
      Configuration conf) throws IOException {
    return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
        JobSubmissionProtocol.versionID, addr,
        UserGroupInformation.getCurrentUser(), conf,
        NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
  }
 
18  

第7行：此时获得一个实现JobSubmissionProtocol 的RPC调用，即JobTracker的代理，jobSubmitClient就是JobSubmissionProtocol接口的实现类。JobSubmissionProtocol是JobClient和Jobtracker通信的桥梁。

在connect方法中创建了jobClient，在submit方法中JobClient通过submitJobInternal(conf)方法正式向JobTracker提交作业。summitJobInternal(conf

)实现如下：

public RunningJob submitJobInternal(final JobConf job
                               ) throws FileNotFoundException, 
                                        ClassNotFoundException,
                                        InterruptedException,
                                        IOException {
    /*
     * configure the command line options correctly on the submitting dfs
     */
    return ugi.doAs(new PrivilegedExceptionAction<RunningJob>() {
      public RunningJob run() throws FileNotFoundException, 
      ClassNotFoundException,
      InterruptedException,
      IOException{
        JobConf jobCopy = job;
        Path jobStagingArea = JobSubmissionFiles.getStagingDir(JobClient.this,
            jobCopy);
        JobID jobId = jobSubmitClient.getNewJobId();
        Path submitJobDir = new Path(jobStagingArea, jobId.toString());
        jobCopy.set("mapreduce.job.dir", submitJobDir.toString());
        JobStatus status = null;
        try {
          populateTokenCache(jobCopy, jobCopy.getCredentials());

          copyAndConfigureFiles(jobCopy, submitJobDir);

          // get delegation token for the dir
          TokenCache.obtainTokensForNamenodes(jobCopy.getCredentials(),
                                              new Path [] {submitJobDir},
                                              jobCopy);

          Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
          int reduces = jobCopy.getNumReduceTasks();
          InetAddress ip = InetAddress.getLocalHost();
          if (ip != null) {
            job.setJobSubmitHostAddress(ip.getHostAddress());
            job.setJobSubmitHostName(ip.getHostName());
          }
          JobContext context = new JobContext(jobCopy, jobId);

          // Check the output specification
          if (reduces == 0 ? jobCopy.getUseNewMapper() : 
            jobCopy.getUseNewReducer()) {
            org.apache.hadoop.mapreduce.OutputFormat<?,?> output =
              ReflectionUtils.newInstance(context.getOutputFormatClass(),
                  jobCopy);
            output.checkOutputSpecs(context);
          } else {
            jobCopy.getOutputFormat().checkOutputSpecs(fs, jobCopy);
          }
          
          jobCopy = (JobConf)context.getConfiguration();

          // Create the splits for the job
          FileSystem fs = submitJobDir.getFileSystem(jobCopy);
          LOG.debug("Creating splits at " + fs.makeQualified(submitJobDir));
          int maps = writeSplits(context, submitJobDir);
          jobCopy.setNumMapTasks(maps);

          // write "queue admins of the queue to which job is being submitted"
          // to job file.
          String queue = jobCopy.getQueueName();
          AccessControlList acl = jobSubmitClient.getQueueAdmins(queue);
          jobCopy.set(QueueManager.toFullPropertyName(queue,
              QueueACL.ADMINISTER_JOBS.getAclName()), acl.getACLString());

          // Write job file to JobTracker's fs        
          FSDataOutputStream out = 
            FileSystem.create(fs, submitJobFile,
                new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));

          // removing jobtoken referrals before copying the jobconf to HDFS
          // as the tasks don't need this setting, actually they may break
          // because of it if present as the referral will point to a
          // different job.
          TokenCache.cleanUpTokenReferral(jobCopy);

          try {
            jobCopy.writeXml(out);
          } finally {
            out.close();
          }
          //
          // 真正开始提交作业
          //
          printTokens(jobId, jobCopy.getCredentials());
       //利用jobSumitClient调用submitJob向JobTracker提交作业
          status = jobSubmitClient.submitJob(
              jobId, submitJobDir.toString(), jobCopy.getCredentials());
          JobProfile prof = jobSubmitClient.getJobProfile(jobId);
          if (status != null && prof != null) {
            return new NetworkedJob(status, prof, jobSubmitClient);
          } else {
            throw new IOException("Could not launch job");
          }
        } finally {
          if (status == null) {
            LOG.info("Cleaning up the staging area " + submitJobDir);
            if (fs != null && submitJobDir != null)
              fs.delete(submitJobDir, true);
          }
        }
      }
    });
  }

 下图描述了作业提交和初始化的过程：

 上图中的1，2表示被调用函数被调用的顺序。

1)关于DistributedCache:

MapReduce作业文件的上传和下载都是由DistributedCache工具完成的。

2)关于复制作业运行所需要的文件到JobTracker所在的文件系统的指定目录(一般是HDFS上的某个目录)

通常而言，对于一个典型的Java MapReduce作业，可能包含以下资源。也就是需要将这些资源复制到HDFS上的某个目录

a.程序jar包：用户用Java编写的MapReduce应用程序jar包。(job.jar)

b.作业配置文件：描述MapReduce应用程序的配置信息(job.xml)

c.依赖的第三方jar包：应用程序依赖的第三方jar包，提交作业时用参数“-libjars”指定。

d.依赖的归档文件：应用程序中用到多个文件，可直接打包成归档文件，提交作业时用参数“-archives”指定。

e.依赖的普通文件：应用程序中可能用到普通文件，比如文本格式的字典文件，提交作业时用参数“-files”指定。

其实，一般复制的文件只有job.jar,job.xml,job.split,job.splitmetainfo .这些统称为作业文件。

 

      作业文件上传到HDFS后，可能会有大量节点同时从HDFS上下载这些文件，进而产生文件访问热点现象，造成性能瓶颈。为此，JobClient上传这些文件时会调高它们的副本数（由参数mapred.submit.replication指定，默认是10）以通过分摊负载方式避免产生访问热点。

再看一下core-site.xml中的一个配置项：

<property>
<name>hadoop.tmp.dir</name>
<value>/usr/local/hadoop/tmp</value>
</property>     

hadoop.tmp.dir 是hadoop文件系统(HDFS)依赖的基础配置，很多路径都依赖它,并且NameNode的元数据备份等信息也会放在此这个目录下，如果不配置，其默认路径是/tmp，这里的/tmp是HDFS中的/tmp，并不是本机Linux中的/tmp,而/tmp是系统的临时目录，系统重启时往往会被清空，所以需要自定义一个持久化的数据目录。

运行一个WordCount作业，当前运行的作业的相关的各种文件在由JobClient提交作业的时候，是复制到了HDFS的某个目录上的，作业运行成功之后会再从HDFS上删除掉。

可以通过两个属性查找到当前运行作业相关的各种文件：

作业属性                                                                                属性值                                                      说明

mapreduce.jobtracker.staging.root.dir                              ${hadoop.tmp.dir}/mapred/staging                  复制过来的作业相关文件在HDFS上的存放位置

mapreduce.job.dir                            ${mapreduce.jobtracker.staging.root.dir}/${user}/.staging/${jobId}   用户{user}的某个作业相关文件的存放位置

而配置文件中hadoop.tmp.dir=/usr/local/hadoop/tmp，很容易找到WordCount作业在HDFS上的相关文件。

                                                                                       

参考：

http://blog.sina.com.cn/s/blog_9d31d3870101dkxt.html