hadoop job初始化源码浅析

时间：2021-09-17 19:37:42

hadoop的job提交过程相对来说还是有点复杂的，所以在学习源码的时候会显得有些乱，时常看了后面忘了前面，所以在看了多遍之后决定用文章的方式记录下来，一边自己下次再看的时候能够清晰些，同时也为初次接触这方面源码的同学提供一些帮助吧。希望自己可以写的足够详细。(本文针对hadoop1.2.1)

1.job.waitForCompletion：一般情况下我们提交一个job都是通过job.waitForCompletion方法提交，该方法内部会调用job.submit()方法

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();
  }

2.job.submit()：在submit中会调用setUseNewAPI()，setUseNewAPI()这个方法主要是判断是使用新的api还是旧的api，之后会调用connect()方法，该方法主要是实例化jobClient，然后会调用jobClient.submitJobInternal(conf)这个方法进行job的提交 

public void submit() throws IOException, InterruptedException, 
                              ClassNotFoundException {
    ensureState(JobState.DEFINE);
    setUseNewAPI();

    // Connect to the JobTracker and submit the job
    connect();
    info = jobClient.submitJobInternal(conf);
    super.setJobID(info.getID());
    state = JobState.RUNNING;
   }

3.jobClient.submitJobInternal()：这个方法会将job运行时所需的所有文件上传到jobTarcker文件系统（一般是hdfs）中，同时进行备份（备份数默认是10，通过mapred.submit.replication变量可以设置），这个方法需要深入进行解读。

4.JobSubmissionFiles.getStagingDir：这个方法是在jobClient.submitJobInternal()最先调用的，这个方法主要是获取一个job提交的根目录，主要是通过Path stagingArea = client.getStagingAreaDir();方法获得，这个方法最终会调用jobTracker.getStagingAreaDirInternal()方法，代码如下：

 private String getStagingAreaDirInternal(String user) throws IOException {
    final Path stagingRootDir =
      new Path(conf.get("mapreduce.jobtracker.staging.root.dir",
            "/tmp/hadoop/mapred/staging"));
    final FileSystem fs = stagingRootDir.getFileSystem(conf);
    return fs.makeQualified(new Path(stagingRootDir,
                              user+"/.staging")).toString();
  }
在获取了stagingDir之后会执行JobID jobId = jobSubmitClient.getNewJobId();为job获取一个jobId，然后执行Path submitJobDir = new Path(jobStagingArea, jobId.toString());获得该job提交的路径，也就是在stagingDir目录下建一个以jobId为文件名的目录。有了submitJobDir之后就可以将job运行所需的全部文件上传到对应的目录下了，具体是调用jobClient.copyAndConfigureFiles(jobCopy, submitJobDir)这个方法。

5.jobClient.copyAndConfigureFiles(jobCopy, submitJobDir)：这个方法最终调用jobClient.copyAndConfigureFiles(job, jobSubmitDir, replication);这个方法实现文件上传。

6.jobClient.copyAndConfigureFiles(job, jobSubmitDir, replication)：这个方法首先获取用户在使用命令执行job的时候所指定的-libjars, -files, -archives文件，对应的conf配置参数是tmpfiles tmpjars tmparchives，这个过程是在ToolRunner.run()的时候进行解析的，当用户指定了这三个参数之后，会将这三个参数对应的文件都上传到hdfs上，下面我们具体看一个参数的处理：tmpfiles（其他两个基本相同）

7.jobClient处理tmpfiles：该方法会将tmpfiles参数值按‘，’分割，然后将每一个文件上传到hdfs，其中如何文件的路径本身就在hdfs中，那么将不进行上传操作，上传操作只针对文件不在hdfs中的文件。调用的方法是：Path newPath = copyRemoteFiles(fs,filesDir, tmp, job, replication)，该方法内部使用的是FileUtil.copy(remoteFs, originalPath, jtFs, newPath, false, job)方法将文件上传至hdfs，注意此处的remoteFs和jtFs，remoteFs就是需上传文件的原始文件系统，jtFs则是jobTracker的文件系统（hdfs）。在文件上传至hdfs之后，会执行DistributedCache.createSymlink(job)这个方法，这个方法是创建一个别名(好像是这么个名字)，这里需要注意的是tmpfiles和tmparchives都会创建别名，而tmpjars则不会，个人认为tmpjars则jar文件，不是用户在job运行期间调用，所以不需要别名，而tmpfiles和tmparchives则在job运行期间用户可能会调用，所以使用别名可以方便用户调用

8.将这三个参数指定的文件上传到hdfs之后，需要将job的jar文件上传到hdfs，名称为submitJobDir/job.jar，使用fs.copyFromLocalFile(originalJarFile, submitJarFile)上传即可。

到这里jobClient.copyAndConfigureFiles(jobCopy, submitJobDir)方法就完成了，期间丢了jobClient.copyAndConfigureFiles(jobCopy, submitJobDir)，TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(job)，TrackerDistributedCacheManager.getDelegationTokens(job,  job.getCredentials())三个方法，这三个方法是进行一些cached archives and files的校验和保存其时间戳和权限内容

9.继续我们的jobClient.submitJobInternal()方法，这之后会根据我们设置的outputFormat类执行output.checkOutputSpecs(context)，进行输出路径的检验，主要是保证输出路径不存在，存在会抛出异常。这之后就是对输入文件进行分片操作了，writeSplits(context, submitJobDir)。

10.jobClient.writeSplits()：这个方法内部会根据我们之前判断的使用new-api还是old-api分别进行分片操作，我们只看new-api的分片操作。

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,
      Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    JobConf jConf = (JobConf)job.getConfiguration();
    int maps;
    if (jConf.getUseNewMapper()) {
      maps = writeNewSplits(job, jobSubmitDir);
    } else {
      maps = writeOldSplits(jConf, jobSubmitDir);
    }
    return maps;
  }
11.jobClient.writeNewSplits()：这个方法主要是根据我们设置的inputFormat.class通过反射获得inputFormat对象，然后调用inputFormat对象的getSplits方法，当获得分片信息之后调用JobSplitWriter.createSplitFiles方法将分片的信息写入到submitJobDir/job.split文件中。 

 private <T extends InputSplit>
  int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    Configuration conf = job.getConfiguration();
    InputFormat<?, ?> input =
      ReflectionUtils.newInstance(job.getInputFormatClass(), conf);

    List<InputSplit> splits = input.getSplits(job);
    T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);

    // sort the splits into order based on size, so that the biggest
    // go first
    Arrays.sort(array, new SplitComparator());
    JobSplitWriter.createSplitFiles(jobSubmitDir, conf,
        jobSubmitDir.getFileSystem(conf), array);
    return array.length;
  }
12.JobSplitWriter.createSplitFiles：这个方法的作用就是讲分片信息写入到submitJobDir/job.split文件中，方法内部调用JobSplitWriter.writeNewSplits进行写操作

13.JobSplitWriter.writeNewSplits：该方法具体对每一个InputSplit对象进行序列化写入到输出流中，具体每个InputSplit对象写入的信息包括：split.getClass().getName()，serializer.serialize(split)将整个对象序列化。然后将InputSplit对象的locations信息放入SplitMetaInfo对象中，同时还包括InputSpilt元信息在job.split文件中的偏移量，该InputSplit的长度，再将SplitMetaInfo对象。然后调用JobSplitWriter.writeJobSplitMetaInfo()方法将SplitMetaInfo对象写入submitJobDir/job.splitmetainfo文件中。

14.JobSplitWriter.writeJobSplitMetaInfo()：将SplitMetaInfo对象写入submitJobDir/job.splitmetainfo文件中，具体写入的信息包括：JobSplit.META_SPLIT_FILE_HEADER，splitVersion，allSplitMetaInfo.length（SplitMetaInfo对象的个数，一个split对应一个SplitMetaInfo），然后分别将所有的SplitMetaInfo对象序列化到输出流中，到此文件的分片工作完成。

15.继续回头看jobClient.submitJobInternal()方法：在上一步进行分片操作之后，或返回切片的数目，据此设定map的数量，所以在job中设置的map数量是没有用的。

16.继续往下走：

String queue = jobCopy.getQueueName();
          AccessControlList acl = jobSubmitClient.getQueueAdmins(queue);
          jobCopy.set(QueueManager.toFullPropertyName(queue,
              QueueACL.ADMINISTER_JOBS.getAclName()), acl.getACLString());
这三句话是获得job对应的任务队列信息，这里涉及到hadoop的作业调度内容，就不深入研究了
17.继续：下面就是讲job的配置文件信息(jobConf对象)写入到xml文件中，以便用户查看，具体文件是：submitJobDir/job.xml，通过jobCopy.writeXml(out)方法，
方法比较简单，就是写xml文件。下面就进入到jobTracker提交任务环节了，status = jobSubmitClient.submitJob(jobId, submitJobDir.toString(), jobCopy.getCredentials())，
就到这吧，后面下次再慢慢研究。
总结下：在用户提交job之后，第一步主要是jobClient对job进行一些必要的文件上传操作，主要包括：
1）为job生成一个jobId，然后获得job提交的stagingDir，根据jobId获得submitJobDir，之后所有的job运行时文件豆浆保存在此目录下
2）将用户在命令行通过-libjars, -files, -archives指定的文件上传到jobTracker的文件系统中，并将job.jar上传到hdfs中
3）校验输出路径
4）进行输入文件的分片操作，并将分片信息写入submitJobDir下的相应文件中，有两个文件：job.split以及job.splitmetainfo
5）将job的配置参数(jobConf对象)写入到job.xml文件中
这就是jobClient提交job的全部过程，如有遗漏下面评论指出，谢谢

标签：

相关文章