JobTracker任务调度器之JobQueueTaskScheduler

谁借莪１个温暖的怀抱￠ 2022-06-17 09:18 230阅读 0赞

在客户端把作业提交给JobTracker节点之后，JobTracker节点就可以使用任务调度器TaskScheduler将这个Job的任务分配给那些合适的TaskTracker节点来执行。当然在JobTracker调度该Job之前，必须要确保该Job的JobInProgress被初始化了，即将Job划分为若干个map任务和reduce任务。在JobTracker中有一个基于优先级的FIFO任务调度器JobQueueTaskScheduler。对于目前的Hadoop任务调度设计来看，它是一个“拉”的过程，即每一个TaskTracker节点主动向JobTracker节点请求作业的任务，而不是当有新作业的时候，JobTracker节点主动给TaskTracker节点分配任务。先来看看它的类图：

![0_1323160465QN4B.gif][]

在上面的类图中，我们可以看出JobQueueTaskScheduler类依赖于两个JobInProgressListener的实现类，其中JobQueueJobInProgressListener类被用来按照优先级队列的方式来管理Job，EagerTaskInitializationListener类被用来初始化Job，即对Job进行map任务和reduce任务的切分。关于这两个JobInProgressListener类的具体实现，我在这里不再作详细的阐述，因为我将会在以后专门进行详细地讨论，有兴趣的童鞋可以自己查先研究它的源代码。所以，当JobTracker给某个TaskTracker分配任务时，它就会调用TaskScheduler的assignTasks(TaskTrackerStatus)方法，让TaskScheduler给该TaskTracker分配任务。那么，究竟TaskScheduler是如何给TaskTracker任务分配任务的，这就得看TaskScheduler的具体实现了，Hadoop允许用户自定义TaskScheduler来根据自己的实际情况来调度Job任务，这个具体实现可以通过配置文件中的mapred.jobtracker.taskScheduler项来设置。TaskScheduler开放给了用户四个可用的方法：

/*开始任务调度器*/ 
      public void start() throws IOException {
        // do nothing
      }
    
      /*关闭任务调度器*/
      public void terminate() throws IOException {
        // do nothing
      }
    
      /*给一个TaskTracker节点分配若干个任务*/
      public abstract List<Task> assignTasks(TaskTrackerStatus taskTracker) throws IOException;
    
      /*获取属于某一个队列的所有JobInProgress*/
      public abstract Collection<JobInProgress> getJobs(String queueName);

下面以JobQueueTaskScheduler为例，来详细的讨论如何给一个TaskTracker分配任务。

![0_1323162482gevz.gif][]

# 1.分配map任务 #

无论怎样，任务调度器JobQueueTaskScheduler总是按照优先级的FIFO来调度每一个Job的Map任务，但对于每一个Job，它只给一次机会，也就是说它顶多只调度该Job的一个Map任务，然后就再调度其它Job的一个Map任务给当前的TaskTracker节点，同时优先分配一个Job的本地任务，当给当前的TaskTracker节点分配了一个非本地任务时，任务调度器JobQueueTaskScheduler就不会再给该TaskTracker节点分配Map任务了，尽管该TaskTracker节点还有空闲的Map Slot。这主要考虑到TaskTracker节点执行非本地任务的代价(CPU不是主要的，关键是网络带宽)。

![0_1323164818WiHI.gif][]

# 2.分配reduce任务 #

![0_1323168220WwwZ.gif][]

给一个计算节点分配map/reduce任务的源代码如下：

/**
       * 给一个计算节点分配若干个任务
       */
      public synchronized List<Task> assignTasks(TaskTrackerStatus taskTracker)throws IOException {
    
        ClusterStatus clusterStatus = taskTrackerManager.getClusterStatus();		
        final int numTaskTrackers = clusterStatus.getTaskTrackers();				//当前集群可用的计算节点数量
        final int clusterMapCapacity = clusterStatus.getMaxMapTasks();				//当前集群执行map任务的总计算能力(Map Slot的总数量)
        final int clusterReduceCapacity = clusterStatus.getMaxReduceTasks();		//当前集群执行reduce任务的总计算能力(Reduce Slot的总数量)
    
        //当前集群待调度的作业队列
        Collection<JobInProgress> jobQueue = jobQueueJobInProgressListener.getJobQueue();
    
        // Get map + reduce counts for the current tracker.
        final int trackerMapCapacity = taskTracker.getMaxMapTasks();				//当前计算节点执行map任务的总计算能力(Map Slot的最大数量)
        final int trackerReduceCapacity = taskTracker.getMaxReduceTasks();		//当前计算节点执行reduce任务的总计算能力(Reduce Slot的最大数量)
        final int trackerRunningMaps = taskTracker.countMapTasks();				//当前计算节点正在使用的map计算能力(正在使用Map Slot执行map任务的数量)
        final int trackerRunningReduces = taskTracker.countReduceTasks();			//当前计算节点正在使用的reduce计算能力(正在使用Reduce Slot执行reduce任务的数量)
    
        //用来存储给该计算节点分配的任务
        List<Task> assignedTasks = new ArrayList<Task>();
    
        int remainingReduceLoad = 0;	//当前集群执行map尚需的计算能力
        int remainingMapLoad = 0;		//当前集群执行reduce尚需的计算能力
        //通过所有正在执行的作业情况，统计当前尚需的计算能力
        synchronized (jobQueue) {
          for (JobInProgress job : jobQueue) {
            if (job.getStatus().getRunState() == JobStatus.RUNNING) {
              remainingMapLoad += (job.desiredMaps() - job.finishedMaps());
                //当前作业是否可以开始调度reduce任务
              if (job.scheduleReduces()) {
                remainingReduceLoad += (job.desiredReduces() - job.finishedReduces());
              }
              
            }
          }
        }
        LOG.debug("There are "+remainingMapLoad+" running/pending map tasks and "+remainingReduceLoad+" running/pending reduce tasks in the cluster now!");
        
    
        //计算当前map任务的负载
        double mapLoadFactor = 0.0;
        if(clusterMapCapacity > 0) {
          mapLoadFactor = (double)remainingMapLoad / clusterMapCapacity;
        }
        //计算当前reduce任务的负载
        double reduceLoadFactor = 0.0;
        if (clusterReduceCapacity > 0) {
          reduceLoadFactor = (double)remainingReduceLoad / clusterReduceCapacity;
        }
            
         //基于当前集群map任务的负载及当前节点节点的总计算能力，估算此时其应该使用的计算能力
        final int trackerCurrentMapCapacity = Math.min((int)Math.ceil(mapLoadFactor * trackerMapCapacity), trackerMapCapacity);
         //计算当前计算节点剩余的计算能力
        int availableMapSlots = trackerCurrentMapCapacity - trackerRunningMaps;
        boolean exceededMapPadding = false;
        if(availableMapSlots > 0) {
          exceededMapPadding = exceededPadding(true, clusterStatus, trackerMapCapacity);
        }
        int numLocalMaps = 0;
        int numNonLocalMaps = 0;
        if(availableMapSlots > 0) LOG.debug("Try to assign "+availableMapSlots+" map tasks to TaskTracker["+taskTracker.trackerName+"]..");
        else LOG.debug("Can not assign map tasks to TaskTracker["+taskTracker.trackerName+"], because it doesn't have any free map slot or it is overloaded.");
        scheduleMaps:
        for (int i=0; i < availableMapSlots; ++i) {
        	
          synchronized (jobQueue) {
            for (JobInProgress job : jobQueue) {
              if (job.getStatus().getRunState() != JobStatus.RUNNING) {
                continue;
              }
              
              Task t = null;
              
               //尝试给当前计算节点分配一个本地map任务
              t = job.obtainNewLocalMapTask(taskTracker, numTaskTrackers,taskTrackerManager.getNumberOfUniqueHosts());
              if(t != null) {
            	  LOG.debug("assign a local map Task["+t.getTaskID()+"] to TaskTracker["+taskTracker.trackerName+"]");
                assignedTasks.add(t);
                ++numLocalMaps;
                
                 //需要当前计算节点保留部分map计算能力，所以结束对当前计算节点的map任务分配
                if(exceededMapPadding) {
                  break scheduleMaps;
                }
    
                break;
              }
              
               //尝试给当前计算节点分配一个非本地map任务，如果分配成功则结束对该节点的map任务分配，
               //以避免它抢走了其它计算节点的本地map任务
              t = job.obtainNewNonLocalMapTask(taskTracker, numTaskTrackers,taskTrackerManager.getNumberOfUniqueHosts());
              if(t != null) {
                assignedTasks.add(t);
                ++numNonLocalMaps;
    
                break scheduleMaps;
              }
            }//for
          }
          
        }//for
        
        int assignedMaps = assignedTasks.size();
    
        //基于当前集群reduce任务的负载及当前节点节点的总计算能力，估算此时其应该使用的reduce计算能力
        final int trackerCurrentReduceCapacity = Math.min((int)Math.ceil(reduceLoadFactor * trackerReduceCapacity), trackerReduceCapacity);
        //计算当前节点剩余的reduce计算能力
        final int availableReduceSlots = Math.min((trackerCurrentReduceCapacity - trackerRunningReduces), 1);
        boolean exceededReducePadding = false;
        if(availableReduceSlots > 0) {
          exceededReducePadding = exceededPadding(false, clusterStatus, trackerReduceCapacity);
           //给当前计算节点至多分配一个reduce任务
          synchronized (jobQueue) {
        	 LOG.debug("try to assign 1 reduce task to TaskTracker["+taskTracker.trackerName+"]..");
            for (JobInProgress job : jobQueue) {
              if (job.getStatus().getRunState() != JobStatus.RUNNING || job.numReduceTasks == 0) {
                continue;
              }
    
               //尝试给当前计算节点分配一个reduce任务
              Task t = job.obtainNewReduceTask(taskTracker, numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts());
              
              //reduce任务分配成功，所以结束对当前计算节点的reduce任务分配
              if(t != null) {
                assignedTasks.add(t);
                break;
              }
              
              //reduce任务分配失败，但需要当前计算节点保留部分reduce计算能力，所以结束对当前计算节点的reduce任务分配
              if(exceededReducePadding) {
                break;
              }
            }//for
            
          }
        }
        
        if (LOG.isDebugEnabled()) {
          LOG.debug("Task assignments for " + taskTracker.getTrackerName() + " --> " +
                    "[" + mapLoadFactor + ", " + trackerMapCapacity + ", " + 
                    trackerCurrentMapCapacity + ", " + trackerRunningMaps + "] -> [" + 
                    (trackerCurrentMapCapacity - trackerRunningMaps) + ", " +
                    assignedMaps + " (" + numLocalMaps + ", " + numNonLocalMaps + 
                    ")] [" + reduceLoadFactor + ", " + trackerReduceCapacity + ", " + 
                    trackerCurrentReduceCapacity + "," + trackerRunningReduces + 
                    "] -> [" + (trackerCurrentReduceCapacity - trackerRunningReduces) + 
                    ", " + (assignedTasks.size()-assignedMaps) + "]");
        }
    
        return assignedTasks;
      }

# 3.计算一个TaskTracker节点是否需要保留部分计算能力(Map/Reduce Slots) #

在任务调度器JobQueueTaskScheduler的实现中，如果在集群中的TaskTracker节点比较多的情况下，它总是会想办法让若干个TaskTracker节点预留一些空闲的slots(计算能力)，以便能够快速的处理优先级比较高的Job的Task或者发生错误的Task，以保证已经被调度的作业的完成。它的具体实现如下：

/**
       * 预留计算能力的集群最小规模
      */
      private static final int MIN_CLUSTER_SIZE_FOR_PADDING = 3;
    
      /**
       * 判断当前集群是否需要预留一部分map/reduce计算能力来执行那些失败的、紧急的或特殊的任务
       */
      private boolean exceededPadding(boolean isMapTask, ClusterStatus clusterStatus, int maxTaskTrackerSlots) { 
    	  
    	 //当前集群可用的计算节点数量
        int numTaskTrackers = clusterStatus.getTaskTrackers();
        //当前集群正在执行的map/reduce任务总数量 
        int totalTasks = (isMapTask) ? clusterStatus.getMapTasks() : clusterStatus.getReduceTasks();
        //当前集群执行的map/reduce任务的总计算能力
        int totalTaskCapacity = isMapTask ? clusterStatus.getMaxMapTasks() : clusterStatus.getMaxReduceTasks();
    
        //当前集群待调度的作业队列
        Collection<JobInProgress> jobQueue = jobQueueJobInProgressListener.getJobQueue();
    
        boolean exceededPadding = false;	//当前集群应该预留的计算能力
        synchronized (jobQueue) {
          int totalNeededTasks = 0;
          for(JobInProgress job : jobQueue) {
            if(job.getStatus().getRunState() != JobStatus.RUNNING || job.numReduceTasks == 0) {
              continue;
            }
    
            //统计当前集群的map/reduce计算需求，计算应该预留多少map/reduce计算力
            totalNeededTasks += isMapTask ? job.desiredMaps() : job.desiredReduces();
            int padding = 0;
            if(numTaskTrackers > MIN_CLUSTER_SIZE_FOR_PADDING) {
              padding = Math.min(maxTaskTrackerSlots, (int) (totalNeededTasks * padFraction));
            }
            
            //当前集群剩余的map/reduce计算能力不足预留的，则让当前计算节点预留一部分map/reduce计算能力
            if(totalTasks + padding >= totalTaskCapacity) {
              exceededPadding = true;
              break;
            }
          }//for
        }
    
        return exceededPadding;
      }

其中，全局变量padFraction的默认值为0.01，也可通过配置文件中的mapred.jobtracker.taskalloc.capacitypad项来设置。还要值得说明的是，任务调度器TaskScheduler只负责调度作业的正式Map/Reduce任务，而作业的其它辅助任务都是交由JobTracker来调度的，如JobSetup、JobCleanup、TaskCleanup任务等。这一点是非常值得用户在自定义任务调度器的时候注意的。对于JobQueueTaskScheduler的任务调度实现原则可总结如下：  
        1.先调度优先级高的作业，统一优先级的作业则先进先出；  
        2.尽量使集群每一个TaskTracker达到负载均衡(这个均衡是task数量上的而不是实际的工作强度)；  
        3.尽量分配作业的本地任务给TaskTracker，但不是尽快分配作业的本地任务给TaskTracker，最多分配一个非本地任务给TaskTracker(一是保证任务的并发性，二是避免有些TaskTracker的本地任务被偷走)，最多分配一个reduce任务；  
         4.为优先级或者紧急的Task预留一定的slot；

[0_1323160465QN4B.gif]: /images/20220617/902417668d9942eb83eb79c79a8b6a53.png
[0_1323162482gevz.gif]: /images/20220617/33617812c6b6405cb1a304e85d10fad8.png
[0_1323164818WiHI.gif]: /images/20220617/1e543505f1b84501b732555536add329.png
[0_1323168220WwwZ.gif]: /images/20220617/d0325bda00ab47069a5595959659f773.png