Hadoop job初始化源码浅析

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运行期间用户可能会调用,所以使用别名可以方便用户调用

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:http://www.heiqu.com/fe8416bb2e7d2017fc006b1fc699b9bf.html