监听器初始化Job、JobTracker相应TaskTracker心跳、调度器分配task源码级分析

  1  /**
  2    * Construct the splits, etc.  This is invoked from an async
  3    * thread so that split-computation doesn‘t block anyone.
  4    */
  5   //任务Task分两种: MapTask 和reduceTask，它们的管理对象都是TaskInProgress 。
  6   public synchronized void initTasks() 
  7   throws IOException, KillInterruptedException, UnknownHostException {
  8     if (tasksInited || isComplete()) {
  9       return;
 10     }
 11     synchronized(jobInitKillStatus){
 12       if(jobInitKillStatus.killed || jobInitKillStatus.initStarted) {
 13         return;
 14       }
 15       jobInitKillStatus.initStarted = true;
 16     }
 17 
 18     LOG.info("Initializing " + jobId);
 19     final long startTimeFinal = this.startTime;
 20     // log job info as the user running the job
 21     try {
 22     userUGI.doAs(new PrivilegedExceptionAction<Object>() {
 23       @Override
 24       public Object run() throws Exception {
 25         JobHistory.JobInfo.logSubmitted(getJobID(), conf, jobFile, 
 26             startTimeFinal, hasRestarted());
 27         return null;
 28       }
 29     });
 30     } catch(InterruptedException ie) {
 31       throw new IOException(ie);
 32     }
 33     
 34     // log the job priority
 35     setPriority(this.priority);
 36     
 37     //
 38     // generate security keys needed by Tasks
 39     //
 40     generateAndStoreTokens();
 41     
 42     //
 43     // read input splits and create a map per a split
 44     //
 45     TaskSplitMetaInfo[] splits = createSplits(jobId);
 46     if (numMapTasks != splits.length) {
 47       throw new IOException("Number of maps in JobConf doesn‘t match number of " +
 48               "recieved splits for job " + jobId + "! " +
 49               "numMapTasks=" + numMapTasks + ", #splits=" + splits.length);
 50     }
 51     numMapTasks = splits.length;//map task的个数就是input split的个数
 52 
 53     // Sanity check the locations so we don‘t create/initialize unnecessary tasks
 54     for (TaskSplitMetaInfo split : splits) {
 55       NetUtils.verifyHostnames(split.getLocations());
 56     }
 57     
 58     jobtracker.getInstrumentation().addWaitingMaps(getJobID(), numMapTasks);
 59     jobtracker.getInstrumentation().addWaitingReduces(getJobID(), numReduceTasks);
 60     this.queueMetrics.addWaitingMaps(getJobID(), numMapTasks);
 61     this.queueMetrics.addWaitingReduces(getJobID(), numReduceTasks);
 62 
 63     maps = new TaskInProgress[numMapTasks]; //为每个map tasks生成一个TaskInProgress来处理一个input split 
 64     for(int i=0; i < numMapTasks; ++i) {
 65       inputLength += splits[i].getInputDataLength();
 66       maps[i] = new TaskInProgress(jobId, jobFile,         //类型是map task
 67                                    splits[i], 
 68                                    jobtracker, conf, this, i, numSlotsPerMap);
 69     }
 70     LOG.info("Input size for job " + jobId + " = " + inputLength
 71         + ". Number of splits = " + splits.length);
 72 
 73     // Set localityWaitFactor before creating cache
 74     localityWaitFactor = 
 75       conf.getFloat(LOCALITY_WAIT_FACTOR, DEFAULT_LOCALITY_WAIT_FACTOR);
 76     /* 对于map task，将其放入nonRunningMapCache，是一个Map<Node,List<TaskInProgress>>，也即对于map task来讲，其将会被分配到其input
 77     split所在的Node上。在此，Node代表一个datanode或者机架或者数据中  心。nonRunningMapCache将在JobTracker向TaskTracker分配map task的 时候使用。
 78     */ 
 79     if (numMapTasks > 0) { 
 80         //通过createCache()方法为这些TaskInProgress对象产生一个未执行任务的Map缓存nonRunningMapCache。
 81         //slave端的TaskTracker向master发送心跳时，就可以直接从这个cache中取任务去执行。
 82       nonRunningMapCache = createCache(splits, maxLevel);
 83     }
 84         
 85     // set the launch time
 86     this.launchTime = jobtracker.getClock().getTime();
 87 
 88     //
 89     // Create reduce tasks
 90     //
 91     //其次JobInProgress会创建Reduce的监控对象，这个比较简单，根据JobConf里指定的Reduce数目创建，
 92     //缺省只创建1个Reduce任务。监控和调度Reduce任务的是TaskInProgress类，不过构造方法有所不同，
 93     //TaskInProgress会根据不同参数分别创建具体的MapTask或者ReduceTask。同样地，
 94     //initTasks()也会通过createCache()方法产生nonRunningReduceCache成员。
 95     this.reduces = new TaskInProgress[numReduceTasks];
 96     for (int i = 0; i < numReduceTasks; i++) {
 97       reduces[i] = new TaskInProgress(jobId, jobFile,     //这是reduce task
 98                                       numMapTasks, i, 
 99                                       jobtracker, conf, this, numSlotsPerReduce);
100       /*reducetask放入nonRunningReduces，其将在JobTracker向TaskTracker分配reduce task的时候使用。*/ 
101       nonRunningReduces.add(reduces[i]);
102     }
103 
104     // Calculate the minimum number of maps to be complete before 
105     // we should start scheduling reduces
106     completedMapsForReduceSlowstart = 
107       (int)Math.ceil(
108           (conf.getFloat("mapred.reduce.slowstart.completed.maps", 
109                          DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) * 
110            numMapTasks));
111     
112     // ... use the same for estimating the total output of all maps
113     resourceEstimator.setThreshhold(completedMapsForReduceSlowstart);
114     
115     // create cleanup two cleanup tips, one map and one reduce.
116   //创建两个cleanup task，一个用来清理map，一个用来清理reduce. 
117     cleanup = new TaskInProgress[2];
118 
119     // cleanup map tip. This map doesn‘t use any splits. Just assign an empty
120     // split.
121     TaskSplitMetaInfo emptySplit = JobSplit.EMPTY_TASK_SPLIT;
122     cleanup[0] = new TaskInProgress(jobId, jobFile, emptySplit, 
123             jobtracker, conf, this, numMapTasks, 1);
124     cleanup[0].setJobCleanupTask();
125 
126     // cleanup reduce tip.
127     cleanup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,
128                        numReduceTasks, jobtracker, conf, this, 1);
129     cleanup[1].setJobCleanupTask();
130 
131     // create two setup tips, one map and one reduce.
132     //创建两个初始化 task，一个初始化map，一个初始化reduce. 
133     setup = new TaskInProgress[2];
134 
135     // setup map tip. This map doesn‘t use any split. Just assign an empty
136     // split.
137     setup[0] = new TaskInProgress(jobId, jobFile, emptySplit, 
138             jobtracker, conf, this, numMapTasks + 1, 1);
139     setup[0].setJobSetupTask();
140 
141     // setup reduce tip.
142     setup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,
143                        numReduceTasks + 1, jobtracker, conf, this, 1);
144     setup[1].setJobSetupTask();
145     
146     synchronized(jobInitKillStatus){
147       jobInitKillStatus.initDone = true;
148       if(jobInitKillStatus.killed) {
149         throw new KillInterruptedException("Job " + jobId + " killed in init");
150       }
151     }
152     //JobInProgress创建完TaskInProgress后，最后构造JobStatus并记录job正在执行中，
153     //然后再调用JobHistory.JobInfo.logInited()记录job的执行日志。
154     tasksInited = true;
155     JobHistory.JobInfo.logInited(profile.getJobID(), this.launchTime, 
156                                  numMapTasks, numReduceTasks);
157     
158    // Log the number of map and reduce tasks
159    LOG.info("Job " + jobId + " initialized successfully with " + numMapTasks
160             + " map tasks and " + numReduceTasks + " reduce tasks.");
161   }

  1 //JobQueueTaskScheduler最重要的方法是assignTasks，他实现了工作调度。
  2   @Override
  3   public synchronized List<Task> assignTasks(TaskTracker taskTracker)
  4       throws IOException {
  5     TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus(); 
  6     ClusterStatus clusterStatus = taskTrackerManager.getClusterStatus();
  7     final int numTaskTrackers = clusterStatus.getTaskTrackers();
  8     final int clusterMapCapacity = clusterStatus.getMaxMapTasks();
  9     final int clusterReduceCapacity = clusterStatus.getMaxReduceTasks();
 10 
 11     Collection<JobInProgress> jobQueue =
 12       jobQueueJobInProgressListener.getJobQueue();
 13     //首先它会检查 TaskTracker 端还可以做多少个 map 和 reduce 任务，将要派发的任务数是否超出这个数，
 14     //是否超出集群的任务平均剩余可负载数。如果都没超出，则为此TaskTracker 分配一个 MapTask 或 ReduceTask 。
 15     //
 16     // Get map + reduce counts for the current tracker.
 17     //
 18     final int trackerMapCapacity = taskTrackerStatus.getMaxMapSlots();
 19     final int trackerReduceCapacity = taskTrackerStatus.getMaxReduceSlots();
 20     final int trackerRunningMaps = taskTrackerStatus.countMapTasks();
 21     final int trackerRunningReduces = taskTrackerStatus.countReduceTasks();
 22 
 23     // Assigned tasks
 24     List<Task> assignedTasks = new ArrayList<Task>();
 25 
 26     //
 27     // Compute (running + pending) map and reduce task numbers across pool
 28     //
 29   //计算剩余的map和reduce的工作量：remaining 
 30     int remainingReduceLoad = 0;
 31     int remainingMapLoad = 0;
 32     synchronized (jobQueue) {
 33       for (JobInProgress job : jobQueue) {
 34         if (job.getStatus().getRunState() == JobStatus.RUNNING) {
 35           remainingMapLoad += (job.desiredMaps() - job.finishedMaps());
 36           if (job.scheduleReduces()) {
 37             remainingReduceLoad += 
 38               (job.desiredReduces() - job.finishedReduces());
 39           }
 40         }
 41       }
 42     }
 43 
 44     // Compute the ‘load factor‘ for maps and reduces
 45     double mapLoadFactor = 0.0;
 46     if (clusterMapCapacity > 0) {
 47       mapLoadFactor = (double)remainingMapLoad / clusterMapCapacity;
 48     }
 49     double reduceLoadFactor = 0.0;
 50     if (clusterReduceCapacity > 0) {
 51       reduceLoadFactor = (double)remainingReduceLoad / clusterReduceCapacity;
 52     }
 53         
 54     //
 55     // In the below steps, we allocate first map tasks (if appropriate),
 56     // and then reduce tasks if appropriate.  We go through all jobs
 57     // in order of job arrival; jobs only get serviced if their 
 58     // predecessors are serviced, too.
 59     //
 60 
 61     //
 62     // We assign tasks to the current taskTracker if the given machine 
 63     // has a workload that‘s less than the maximum load of that kind of
 64     // task.
 65     // However, if the cluster is close to getting loaded i.e. we don‘t
 66     // have enough _padding_ for speculative executions etc., we only 
 67     // schedule the "highest priority" task i.e. the task from the job 
 68     // with the highest priority.
 69     //
 70     
 71     final int trackerCurrentMapCapacity = 
 72       Math.min((int)Math.ceil(mapLoadFactor * trackerMapCapacity), 
 73                               trackerMapCapacity);
 74     int availableMapSlots = trackerCurrentMapCapacity - trackerRunningMaps;
 75     boolean exceededMapPadding = false;
 76     if (availableMapSlots > 0) {
 77       exceededMapPadding = 
 78         exceededPadding(true, clusterStatus, trackerMapCapacity);
 79     }
 80     int numLocalMaps = 0;
 81     int numNonLocalMaps = 0;
 82     scheduleMaps:
 83     for (int i=0; i < availableMapSlots; ++i) {
 84       synchronized (jobQueue) {
 85         for (JobInProgress job : jobQueue) {
 86           if (job.getStatus().getRunState() != JobStatus.RUNNING) {
 87             continue;
 88           }
 89 
 90           Task t = null;
 91           
 92           // Try to schedule a node-local or rack-local Map task
 93           t = 
 94             job.obtainNewNodeOrRackLocalMapTask(taskTrackerStatus, 
 95                 numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts());
 96           if (t != null) {
 97             assignedTasks.add(t);
 98             ++numLocalMaps;
 99             
100             // Don‘t assign map tasks to the hilt!
101             // Leave some free slots in the cluster for future task-failures,
102             // speculative tasks etc. beyond the highest priority job
103             if (exceededMapPadding) {
104               break scheduleMaps;
105             }
106            
107             // Try all jobs again for the next Map task 
108             break;
109           }
110           
111           // Try to schedule a node-local or rack-local Map task
112           //产生 Map 任务使用 JobInProgress 的obtainNewMapTask() 方法，
113           //实质上最后调用了 JobInProgress 的 findNewMapTask() 访问nonRunningMapCache 。
114           t = 
115             job.obtainNewNonLocalMapTask(taskTrackerStatus, numTaskTrackers,
116                                    taskTrackerManager.getNumberOfUniqueHosts());
117           
118           if (t != null) {
119             assignedTasks.add(t);
120             ++numNonLocalMaps;
121             
122             // We assign at most 1 off-switch or speculative task
123             // This is to prevent TaskTrackers from stealing local-tasks
124             // from other TaskTrackers.
125             break scheduleMaps;
126           }
127         }
128       }
129     }
130     int assignedMaps = assignedTasks.size();
131 
132     //
133     // Same thing, but for reduce tasks
134     // However we _never_ assign more than 1 reduce task per heartbeat
135     ////分配完map task，再分配reduce task 
136     final int trackerCurrentReduceCapacity = 
137       Math.min((int)Math.ceil(reduceLoadFactor * trackerReduceCapacity), 
138                trackerReduceCapacity);
139     final int availableReduceSlots = 
140       Math.min((trackerCurrentReduceCapacity - trackerRunningReduces), 1);
141     boolean exceededReducePadding = false;
142     if (availableReduceSlots > 0) {
143       exceededReducePadding = exceededPadding(false, clusterStatus, 
144                                               trackerReduceCapacity);
145       synchronized (jobQueue) {
146         for (JobInProgress job : jobQueue) {
147           if (job.getStatus().getRunState() != JobStatus.RUNNING ||
148               job.numReduceTasks == 0) {
149             continue;
150           }
151           //使用JobInProgress.obtainNewReduceTask() 方法，
152           //实质上最后调用了JobInProgress的 findNewReduceTask() 访问 nonRuningReduceCache
153           Task t = 
154             job.obtainNewReduceTask(taskTrackerStatus, numTaskTrackers, 
155                                     taskTrackerManager.getNumberOfUniqueHosts()
156                                     );
157           if (t != null) {
158             assignedTasks.add(t);
159             break;
160           }
161           
162           // Don‘t assign reduce tasks to the hilt!
163           // Leave some free slots in the cluster for future task-failures,
164           // speculative tasks etc. beyond the highest priority job
165           if (exceededReducePadding) {
166             break;
167           }
168         }
169       }
170     }
171     
172     if (LOG.isDebugEnabled()) {
173       LOG.debug("Task assignments for " + taskTrackerStatus.getTrackerName() + " --> " +
174                 "[" + mapLoadFactor + ", " + trackerMapCapacity + ", " + 
175                 trackerCurrentMapCapacity + ", " + trackerRunningMaps + "] -> [" + 
176                 (trackerCurrentMapCapacity - trackerRunningMaps) + ", " +
177                 assignedMaps + " (" + numLocalMaps + ", " + numNonLocalMaps + 
178                 ")] [" + reduceLoadFactor + ", " + trackerReduceCapacity + ", " + 
179                 trackerCurrentReduceCapacity + "," + trackerRunningReduces + 
180                 "] -> [" + (trackerCurrentReduceCapacity - trackerRunningReduces) + 
181                 ", " + (assignedTasks.size()-assignedMaps) + "]");
182     }
183 
184     return assignedTasks;
185   }