hadoop-map-reduce执行流程调研报告

hadoop-map-reduce执行流程调研报告
参与者:
Client       简称:C
JobTracker   简称:JT
TaskTracker  简称:TT
DataNode   简称:DN
NameNode  简称:NN
流程描述:
简述:
作业的提交 -> Map任务的分配和执行 -> Reduce任务的分配和执行 -> 作业的完成
每个具体的任务又分: 准备输入-> 算法的执行 -> 输出结果,三个步骤.
作业的提交:
     1. /home/hadoop/hadoop-1.0.3/bin/hadoo jar  ~/learning-hadoop.jar  com.easou.mr.bookMark.BookMarkMR
     2. RunJar.java  -> main
     3.反射 -BookMarkMR->main
     4. org.apache.hadoop.mapreduce.Job ->waitForCompletion
//提交
i f (state == JobState.DEFINE) {
submit();
}
     5. org.apache.hadoop.mapred.JobClient -> submitJobInternal
     6.
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.jar(执行jar)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.split(输入分片文件)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.splitmetainfo(分片meta)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.xml(jobconf配置信息)
7.
this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);

JobID jobId = jobSubmitClient.getNewJobId();//jobid的生成方式

public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts) throws IOException

synchronized (jobs) {
      synchronized (taskScheduler) {
        jobs.put(job.getProfile().getJobID(), job);
        for (JobInProgressListener listener : jobInProgressListeners) {
          listener.jobAdded(job);
        }
      }
    }
Map reduce任务的分配和执行:
C rpc 调用 addJob 触发监听:
JobQueueJobInProgressListener(用于监控job的运行状态)

EagerTaskInitializationListener;(用于对Job进行初始化)

public void run() {
      JobInProgress job = null;
      while (true) {
        try {
          synchronized (jobInitQueue) {
            while (jobInitQueue.isEmpty()) {
              jobInitQueue.wait();
            }
            job = jobInitQueue.remove(0);
          }
          threadPool.execute(new InitJob(job));
        } catch (InterruptedException t) {
          LOG.info("JobInitManagerThread interrupted.");
          break;
        }
      }
      LOG.info("Shutting down thread pool");
      threadPool.shutdownNow();
}
    //后台守护运行
jobInitManagerThread.setDaemon(true);
    this.jobInitManagerThread.start();
死循环等待job加入,并初始化.


JT初始化task
JobInProgress -> initTasks  
1.从HDFS中读取job.split文件从而生成input splits
2.map task的个数就是input split的个数
numMapTasks = splits.length;
3.为每个map tasks生成一个TaskInProgress来处理一个input split
4.对于map task，将其放入nonRunningMapCache，是一个Map<Node, List<TaskInProgress>>，也即对于map task来讲，其将会被分配到其input split所在的Node上。nonRunningMapCache将在JobTracker向TaskTracker分配map task的时候使用。
5.创建reduce task
6.reduce task放入nonRunningReduces，其将在JobTracker向TaskTracker分配reduce task的时候使用。
6.创建两个cleanup task，一个用来清理map，一个用来清理reduce.
7.创建两个初始化 task，一个初始化map，一个初始化reduce.

TT:
1. 一直运行
2. 每隔一段时间就向JobTracker发送heartbeat
3. 在heartbeat中要返回给JobTracker一些统计信息, 报告给JobTracker，此TaskTracker的当前状态
4. 发送Heartbeat到JobTracker，得到response
5. 从Response中得到此TaskTracker需要做的事情
6. 如果是运行一个新的Task，则将Action添加到任务队列中
7. 当满足下面的条件的时候，此TaskTracker请求JobTracker为其分配一个新的Task来运行：
当前TaskTracker正在运行的map task的个数小于可以运行的map task的最大个数
当前TaskTracker正在运行的reduce task的个数小于可以运行的reduce task的最大个数
8. 向JobTracker发送RPC调用Heartbeat

JT 分配 task:


根据数据所处的位置与Task Tracker的距离，有如下几种data locality级别：
0     node-local    输入分片就在Task Tracker本地
1     rack-local     输入分片在Task Tracker所在的rack内其它Task Tracker上
2     off-switch    输入分片在其它的rack内

  1.先调度优先级高的作业，统一优先级的作业则先进先出；
  2.尽量使集群每一个TaskTracker达到负载均衡(这个均衡是task数量上的而不是实际的工作强度)；
  3.尽量分配作业的本地任务给TaskTracker，但不是尽快分配作业的本地任务给TaskTracker，最多分配一个非本地任务给TaskTracker(一是保证任务的并发性，二是避免有些TaskTracker的本地任务被偷走)，最多分配一个reduce任务；
  4..为紧急的Task预留一定的slot；

作业的完成:
Map Clean Up Task（Map任务服务器的清理任务，用于清理相关的过期的文件和环境...） -> Map Setup Task（Map任务服务器的安装任务，负责配置好相关的环境...） -> Map Tasks -> Reduce Clean Up Task -> Reduce Setup Task -> Reduce Tasks

流程图:
整体分配流程:







Map的具体流程





Reduce的具体流程