以官方Hadoop中的 WordCount案例分析 ，Job作业的提交过程：

public static void main(String[] args) throws Exception {
          // Create a new Job
          Configuration conf=new Configuration(true);
          Job job = Job.getInstance(conf);
          job.setJarByClass(MyWorkCountJob.class);
         // Specify various job-specific parameters    
          job.setJobName("myWorkCountjob");
          //设置输入文件路径
          FileInputFormat.addInputPath(job, new Path("/user/root/hello.txt"));
          //设置输出文件路径
          Path outPath=new Path("/sxt/mr/output");
          if(FileSystem.get(conf).exists(outPath))
           FileSystem.get(conf).delete(outPath);
          FileOutputFormat.setOutputPath(job, outPath);
         job.setMapperClass(MyMapper.class);
          
          job.setReducerClass(MyReducer.class);
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(IntWritable.class);
          // Submit the job, then poll for progress until the job is complete
          job.waitForCompletion(true);//job 提交的入口
     }

waitForCompletion方法

public boolean waitForCompletion(boolean verbose
                                   ) throws IOException, InterruptedException,
                                            ClassNotFoundException {
    if (state == JobState.DEFINE) {
      submit();// 任务提交1.1
    }
    if (verbose) {
      monitorAndPrintJob();//实时监控Job任务并打印相关的日志
    } else {
      // get the completion poll interval from the client.
      int completionPollIntervalMillis =
        Job.getCompletionPollInterval(cluster.getConf());
      while (!isComplete()) {
        try {
          Thread.sleep(completionPollIntervalMillis);
        } catch (InterruptedException ie) {
        }
      }
    }
    return isSuccessful();
  }

1.1 submit 方法

public void submit()
         throws IOException, InterruptedException, ClassNotFoundException {
    ensureState(JobState.DEFINE);//确定job状态
    setUseNewAPI();//默认使用新的API
    connect();//获得与集群的连接
    final JobSubmitter submitter =
        getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
    status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
      public JobStatus run() throws IOException, InterruptedException,
      ClassNotFoundException {
            //异步调用submitJobInternal方法提交任务 1.2
        return submitter.submitJobInternal(Job.this, cluster);
      }
    });
    state = JobState.RUNNING;
    LOG.info("The url to track the job: " + getTrackingURL());
   }

submit方法首先创建了JobSubmitter实例，然后异步调用了JobSubmitter的submitJobInternal方法

1.2 submitJobInternal 方法

JobStatus submitJobInternal(Job job, Cluster cluster)
  throws ClassNotFoundException, InterruptedException, IOException {

    //检查job的输出路径是否存在，如果存在则抛出异常
    checkSpecs(job);
    Configuration conf = job.getConfiguration();
    addMRFrameworkToDistributedCache(conf);

      //初始化临时目录和返回的输出路径。
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
    //configure the command line options correctly on the submitting dfs
    InetAddress ip = InetAddress.getLocalHost();
    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);
      // 获取提交目录
    Path submitJobDir = new Path(jobStagingArea, jobId.toString());
    ......
        //把作业上传到集群中去
      copyAndConfigureFiles(job, submitJobDir);

      Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
     
      // 创建切片列表 找出每个文件的切片列表 合并切片列表的数量就是Map任务个数  客户端统计
      int maps = writeSplits(job, submitJobDir); //2.1核心方法
      conf.setInt(MRJobConfig.NUM_MAPS, maps);//文件分片的大小 就是Map任务数量
      ......

      // Write job file to submit dir    相关配置写入到job.xml中
      writeConf(conf, submitJobFile);
     
      // Now, actually submit the job (using the submit name) 真正的提交作业
      status = submitClient.submitJob( //2.3 提交job到RecourceManager
          jobId, submitJobDir.toString(), job.getCredentials());
     ...
  }

2.1 文件切片操作 writeSplits -> writeNewSplits 计算向数据移动模型的核心

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());
      //将split信息和SplitMetaInfo都写入HDFS中
    JobSplitWriter.createSplitFiles(jobSubmitDir, conf,
        jobSubmitDir.getFileSystem(conf), array);
    return array.length;
  }

writeNewSplits方法中，划分任务数量最关键的代码即为InputFormat的getSplits方法（InputFormat有不同实现类 框架默认的是TextInputFormat）。此时的Input即为TextInputFormat的父类FileInputFormat，其getSplits方法的实现如下：

 public List<InputSplit> getSplits(JobContext job) throws IOException {
    Stopwatch sw = new Stopwatch().start();
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));//默认最小值 1
    long maxSize = getMaxSplitSize(job);//默认最大值 Long类型的最大值

    // generate splits
    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;
            //获取文件的block块列表
        if (file instanceof LocatedFileStatus) {
          blkLocations = ((LocatedFileStatus) file).getBlockLocations();
        } else {
          FileSystem fs = path.getFileSystem(job.getConfiguration());
          blkLocations = fs.getFileBlockLocations(file, 0, length);
        }
        if (isSplitable(job, path))
          long blockSize = file.getBlockSize();
          long splitSize = computeSplitSize(blockSize, minSize, maxSize);
        //核心代码块
          long bytesRemaining = length;
          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;
          }
//核心代码块结束
          if (bytesRemaining != 0) { 
            int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                       blkLocations[blkIndex].getHosts(),
                       blkLocations[blkIndex].getCachedHosts()));
          }
        } else { // not splitable
          splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
                      blkLocations[0].getCachedHosts()));
        }
      } else {
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    // Save the number of input files for metrics/loadgen
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", TimeTaken: " + sw.elapsedMillis());
    }
    return splits;
  }

2.2 核心代码块分析
对每个输入文件进行split划分。注意这只是个逻辑的划分 因此执行的是FileInputFormat类中的getSplits方法。只有非压缩的文件和几种特定压缩方式压缩后的文件才分片。分片的大小由如下几个参数决定：mapreduce.input.fileinputformat.split.maxsize、mapreduce.input.fileinputformat.split.minsize、文件的blocksize大小确定。
具体计算方式为：
Math.max(minSize, Math.min(maxSize, blockSize))
分片的大小有可能比默认块大小64M要大，当然也有可能小于它，默认情况下分片大小为当前HDFS的块大小，64M

第一步 将bytesRemaining(剩余未分片字节数)初始化设置为整个文件的长度
第二步 如果bytesRemaining超过分片大小splitSize一定量才会将文件分成多个InputSplit，SPLIT_SLOP(默认1.1)。接着就会执行如下方法获取block的索引，其中第二个参数是这个block在整个文件中的偏移量

protected int getBlockIndex(BlockLocation[] blkLocations,
                              long offset) {
    for (int i = 0 ; i < blkLocations.length; i++) {
      // is the offset inside this block? 核心代码块 判断当前的偏移量是否在某个block中 是就返回当前index 位置信息
      if ((blkLocations[i].getOffset() <= offset) &&
          (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())){
        return i;
      }
    }
    BlockLocation last = blkLocations[blkLocations.length -1];
    long fileLength = last.getOffset() + last.getLength() -1;
    throw new IllegalArgumentException("Offset " + offset +
                                       " is outside of file (0.." +
                                       fileLength + ")");
  }

第三步 将符合条件的块的索引对应的block信息的主机节点以及文件的路径名、开始的偏移量、分片大小splitSize封装到一个InputSplit中加入List<InputSplit> splits 列表。

第四步 bytesRemaining -= splitSize修改剩余字节大小 循环以上操作 直到不满足条件 剩余bytesRemaining还不为0，表示还有未分配的数据，将剩余的数据及最后一个block加入splits列表

以上是 整个getSplits获取切片的过程。当使用基于FileInputFormat实现InputFormat时，为了提高MapTask的数据本地化，应尽量使InputSplit大小与block大小相同

2.3 submitter 实现了ClientProtocol接口的类 在1.1中connect()连接集群时 调用init初始化方法 由框架读取 HDFS的配置文件中配置了mapreduce.framework.name属性为“yarn”的话，会创建一个YARNRunner对象 submitter 就是YARNRunner 对象
submitter.submitJobInternal(Job.this, cluster)

YARNRunner的构造方法：

public YARNRunner(Configuration conf, ResourceMgrDelegate resMgrDelegate,
     ClientCache clientCache) {
   this.conf = conf;
   try {
     this.resMgrDelegate = resMgrDelegate;
     this.clientCache = clientCache;
     this.defaultFileContext = FileContext.getFileContext(this.conf);
   } catch (UnsupportedFileSystemException ufe) {
     throw new RuntimeException("Error in instantiating YarnClient", ufe);
   }
 }

ResourceMgrDelegate实际上ResourceManager的代理类，其实现了YarnClient接口，通过ApplicationClientProtocol代理直接向RM提交Job，杀死Job，查询Job运行状态等操作。
YarnRunner 类的submitJob方法

public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts)
  throws IOException, InterruptedException {
    addHistoryToken(ts);
    // Construct necessary information to start the MR AM
       //Client构造ASC。ASC中包括了调度队列，优先级，用户认证信息，除了这些基本的信息之外，还包括用来启动AM的CLC信息，一个CLC中包括jar包、依赖文件、安全token，以及运行任务过程中需要的其他文件
    ApplicationSubmissionContext appContext =
      createApplicationSubmissionContext(conf, jobSubmitDir, ts);
    // Submit to ResourceManager
    try {
      ApplicationId applicationId =
          resMgrDelegate.submitApplication(appContext); // 2.4 提交ASC到RecoureManeger 

      ApplicationReport appMaster = resMgrDelegate
          .getApplicationReport(applicationId);
      String diagnostics =
          (appMaster == null ?
              "application report is null" : appMaster.getDiagnostics());
      if (appMaster == null
          || appMaster.getYarnApplicationState() == YarnApplicationState.FAILED
          || appMaster.getYarnApplicationState() == YarnApplicationState.KILLED) {
        throw new IOException("Failed to run job : " +
            diagnostics);
      }
      return clientCache.getClient(jobId).getJobStatus(jobId);
    } catch (YarnException e) {
      throw new IOException(e);
    }
  }

2.4 到这里一个Client就完成了一次Job任务的提交

2.5 YARN 框架 统一的资源管理 任务调度

yarn.png

**NodeManager ** 每个DN上都会对应一个NM进程

• 与RM汇报资源的使用情况
• 管理运行的Container生命周期
  Container：【节点NM上CPU,MEM,I/O大小等资源的虚拟描述】

MR-ApplicationMaster-Container
每个Job作业对应一个AM，避免单点故障，负载到不同的节点
创建Task时需要和RM申请资源（Container），然后向存放具体资源的DN通信，由DN创建Container并且启动进程同时下发任务（这里就实现了计算向数据移动）

Task-Container 任务执行进程
DN上执行的JVM进程，接收到AM下发的任务后，通过反射机制创建具体的任务对象后 执行具体的任务

** 执行流程**
1 RM 在空闲的DN 上启动AM
2 AM向RM申请资源 ,RM将资源分配信息给AM
3 AM在和数据所在的NM节点通信,创建Container并且通知NM启动Container(JVM进程),分发具体任务到NM上，Container通过反射调起具体的任务类执行
4 如果是MapReduce框架 则进入到MapTask流程 具体分析见 http://liujiacai.net/blog/2014/09/07/yarn-intro/