MapReduce｜MR程序如何提交到Yarn集群

hadoop jar example.jar WordCount /file/input  /file/output

elif [ "$COMMAND" = "jar" ] ; then
  CLASS=org.apache.hadoop.util.RunJar

public static void main(String[] args) throws Throwable {
  //构建一个RunJar对象并执行run方法
  new RunJar().run(args);
}

public void run(String[] args) throws Throwable {
  String usage = "RunJar jarFile [mainClass] args...";
  ...
  //获取到Jar文件，即提交的example.jar文件
  JarFile jarFile;
  try {
    jarFile = new JarFile(fileName);
  } catch(IOException io) {
    throw new IOException("Error opening job jar: " + fileName)
      .initCause(io);
  }


  //拿到Jar文件的Main Class，如WordCount.class
  Manifest manifest = jarFile.getManifest();
  if (manifest != null) {
    mainClassName = manifest.getMainAttributes().getValue("Main-Class");
  }
  jarFile.close();
  ...
  mainClassName = mainClassName.replaceAll("/", ".");


  File tmpDir = new File(System.getProperty("java.io.tmpdir"));
  ensureDirectory(tmpDir);
  //声明一个临时文件
  final File workDir;
  try { 
    workDir = File.createTempFile("hadoop-unjar", "", tmpDir);
  } catch (IOException ioe) {
    ...
  }
  ...
  //解压Jar文件
  unJar(file, workDir);


  ClassLoader loader = createClassLoader(file, workDir);
  Thread.currentThread().setContextClassLoader(loader);


  //通过反射拿到MainClass实例
  Class<?> mainClass = Class.forName(mainClassName, true, loader);


  //拿到main方法的实例
  Method main = mainClass.getMethod("main", new Class[] {
    Array.newInstance(String.class, 0).getClass()
  });
  String[] newArgs = Arrays.asList(args)
    .subList(firstArg, args.length).toArray(new String[0]);
  try {
    //通过反射调用main方法
    main.invoke(null, new Object[] { newArgs });
  } catch (InvocationTargetException e) {
    throw e.getTargetException();
  }
  /**
   * TODO 到此为止，跳转到Driver的main方法，也就是自己编写的应用程序的main方法
   *  最后调用Job.waitForCompletion等待应用程序执行完毕
   */
}

public boolean waitForCompletion(boolean verbose
                                 ) throws IOException, InterruptedException,
                                          ClassNotFoundException {
  if (state == JobState.DEFINE) {
    //TODO 提交任务
    submit();
  }
  //打印执行的进度信息
  ...
  return isSuccessful();
}

public void submit() 
       throws IOException, InterruptedException, ClassNotFoundException {
  //1.设置当前Job的状态为DEFINE
  ensureState(JobState.DEFINE);
  //2.设置启用NewAPI，2.x之后属于NewAPI
  setUseNewAPI();
  //3.获取提交客户端，连接Yarn集群，获取到ResourceManager的代理对象
  connect();
  //4.获取一个提交器
  //这里的cluster就是上面初始化的Cluster对象
  //cluster.getFileSystem() ==> HDFS
  //cluster.getClient()     ==> YARNRunner
  final JobSubmitter submitter = 
      getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
  status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
    public JobStatus run() throws IOException, InterruptedException, 
    ClassNotFoundException {
      //调用提交器的submitJobInternal方法执行Job的提交
      return submitter.submitJobInternal(Job.this, cluster);
    }
  });
  //5.提交之后，设置MR的状态为RUNNING
  state = JobState.RUNNING;
  LOG.info("The url to track the job: " + getTrackingURL());
 }

private synchronized void connect()
        throws IOException, InterruptedException, ClassNotFoundException {
  if (cluster == null) {
    cluster = 
      ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
                 public Cluster run()
                        throws IOException, InterruptedException, 
                               ClassNotFoundException {
                   //初始化了Cluster对象
                   return new Cluster(getConfiguration());
                 }
               });
  }
}

private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
    throws IOException {


    synchronized (frameworkLoader) {
      for (ClientProtocolProvider provider : frameworkLoader) {
        LOG.debug("Trying ClientProtocolProvider : "
                + provider.getClass().getName());
        ClientProtocol clientProtocol = null;
        try {
          if (jobTrackAddr == null) {
            //local模式
            //provider=LocalClientProtocolProvider
            //clientProtocol=LocalJobRunner
            clientProtocol = provider.create(conf);
          } else {
            //Yarn模式
            //provider=YarnClientProtocolProvider
            //clientProtocal=YARNRunner
            clientProtocol = provider.create(jobTrackAddr, conf);
          }
    ...
}

public YARNRunner(Configuration conf) {


 this(conf, new ResourceMgrDelegate(new YarnConfiguration(conf)));
}

public ResourceMgrDelegate(YarnConfiguration conf) {
  super(ResourceMgrDelegate.class.getName());
  this.conf = conf;
  //创建一个YarnClient，实现类是YarnClientImpl
  this.client = YarnClient.createYarnClient();
  init(conf);
  start();
}

//ResourceManager的代理对象
protected ApplicationClientProtocol rmClient;

//验证作业的输出路径是否存在，如果存在会报错，
//正常情况是我们进行了配置，但是该路径在作业提交时是不存在的
checkSpecs(job);


//添加应用框架的路径到分布式缓存中
//DistributedCache：把应用的一些资源添加到分布式缓存中，那么程序在执行的时候，
//无论有多少个节点启动任务，这些节点都会自动把分布式缓存中的各种信息
//（小数据文件，配置信息...）同步到节点本地
addMRFrameworkToDistributedCache(conf);


//通过静态方法getStagingDir()获取作业执行时相关资源的存放路径
//默认是/tmp/hadoop-yarn/staging/提交作业用户名{$user}/.staging
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);


//TODO 记录作业提交的主机IP，主机名，并设置配置信息
if (ip != null) {
  submitHostAddress = ip.getHostAddress();
  submitHostName = ip.getHostName();
  conf.set(MRJobConfig.JOB_SUBMITHOST,submitHostName);
  conf.set(MRJobConfig.JOB_SUBMITHOSTADDR,submitHostAddress);
}


//生成JobID
JobID jobId = submitClient.getNewJobID();
job.setJobID(jobId);


//构造提交作业的路径，jobStagingArea 后面拼接 jobID
//当提交一个 Job时，YARN的客户端会把该Job的一切要用的资源初始化并且存储在HDFS
//的工作目录中，以后哪个节点要执行 Task，就从HDFS 目录中拉取资源文件
//主要包括以下三类资源：
//1.xxx.jar
//2.job.xml
//3.shell启动命令
Path submitJobDir = new Path(jobStagingArea, jobId.toString());


try {
    //将jar文件和配置文件上传到上面获取的资源的提交目录 submitJobDir
    copyAndConfigureFiles(job, submitJobDir);


    //上传完成后获取完整的配置文件所在路径,也就是job.xml的路径
    Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);


    ...

    ...
    //获取逻辑切片的数量
    int maps = writeSplits(job, submitJobDir);
    //配置需要启动的 Map Task 的个数
    conf.setInt(MRJobConfig.NUM_MAPS, maps);
    ...

private <T extends InputSplit>
int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
    InterruptedException, ClassNotFoundException {
  Configuration conf = job.getConfiguration();
  //TODO 根据配置的InputFormat类，通过反射创建对应的实例对象
  // 默认为TextInputFormat，是FileInputFormat的子类
  InputFormat<?, ?> input =
    ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
  //TODO 核心方法，InputSplit就是切片对象，getSplits() 方法返回切片对象的集合
  // 这里的getSplits()在FileInputFormat中实现，TextInputFormat并未进行重写
  List<InputSplit> splits = input.getSplits(job);
  //将集合转为数组
  T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);


  //对数组中的元素根据切片大小进行排序
  Arrays.sort(array, new SplitComparator());
  //TODO 将逻辑切片信息写入Job提交的HDFS目录中
  JobSplitWriter.createSplitFiles(jobSubmitDir, conf, 
      jobSubmitDir.getFileSystem(conf), array);
  //返回切片数量
  return array.length;
}

public List<InputSplit> getSplits(JobContext job) throws IOException {
  ...
  //初始化切片对象的集合容器
  List<InputSplit> splits = new ArrayList<InputSplit>();
  List<FileStatus> files = listStatus(job);
  //遍历输入文件
  for (FileStatus file: files) {
    Path path = file.getPath();
    //拿到文件的大小
    long length = file.getLen();
    if (length != 0) {
      BlockLocation[] blkLocations;
      if (file instanceof LocatedFileStatus) {
        blkLocations = ((LocatedFileStatus) file).getBlockLocations();
      } else {
        FileSystem fs = path.getFileSystem(job.getConfiguration());
        blkLocations = fs.getFileBlockLocations(file, 0, length);
      }
      //TODO 判断文件是否可以进行切分
      // 如果可以进行切分
      if (isSplitable(job, path)) {
        //获取配置的块大小
        long blockSize = file.getBlockSize();
        //计算逻辑切片的大小
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);


        //bytesRemaining => 文件剩余大小
        long bytesRemaining = length;
        //TODO 判断文件的剩余大小是否大于切片大小的 1.1 倍
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          //进行切片
          splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                      blkLocations[blkIndex].getHosts(),
                      blkLocations[blkIndex].getCachedHosts()));
          //更新剩余文件大小
          bytesRemaining -= splitSize;
        }
        //TODO 如果剩余大小小于切片大小的 1.1 倍，且剩余大小不为0，那么将剩余的文件作为一个逻辑切片
        if (bytesRemaining != 0) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                     blkLocations[blkIndex].getHosts(),
                     blkLocations[blkIndex].getCachedHosts()));
        }
        //TODO 如果无法切分，将文件整体作为一个逻辑切片
      } else { // not splitable
        splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
                    blkLocations[0].getCachedHosts()));
      }
      //如果文件大小为0，不需要进行切片，添加一个空的逻辑切片
    } else { 
      //Create empty hosts array for zero length files
      splits.add(makeSplit(path, 0, length, new String[0]));
    }
  }
      ...
  }
  return splits;
}

//将解析该job生成的配置文件，写入到HDFS上
writeConf(conf, submitJobFile);

//这里的 submitClient 为 YARNRunner 实例
status = submitClient.submitJob(
    jobId, submitJobDir.toString(), job.getCredentials());

public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts) throws IOException, InterruptedException {
    ...
    try {
      //TODO 提交应用并获取 ApplicationId
      ApplicationId applicationId =
          resMgrDelegate.submitApplication(appContext);
      ...
    }
    ...
}