前言 :本文旨在理清在Hadoop中一个MapReduce作业(Job)在提交到框架后的整个生命周期过程,权作总结和日后参考,如有问题,请不吝赐教。本文不涉及Hadoop的架构设计,如有兴趣请参考相关书籍和文献。在梳 理过程中,我对一些感兴趣的源码也会逐行研究学习,以期强化基础。
作者 :Jaytalent
开始日期 :2013年9月9日参考资料:【1】《Hadoop技术内幕--深入解析MapReduce架构设计与实现原理》董西成【2】 Hadoop 1.0.0 源码
【3】《Hadoop技术内幕--深入解析Hadoop Common和HDFS架构设计与实现原理》蔡斌 陈湘萍
继续
上一篇文章的话题,说说调度器的任务选择机制。
一个MapReduce作业的生命周期大体分为5个阶段
【1】:
1. 作业提交与初始化
2.
任务调度与监控
3. 任务运行环境准备
4. 任务执行
5. 作业完成
当JobTracker收到了来自TaskTracker的心跳后,是如何选择任务的呢?是通过assignTasks方法。下面详细分析该方法。在分析之前,首先提一下Hadoop的调度器调度模型。通常情况下,Hadoop会以队列为单位管理作业和资源。有了队列就产生所谓三级调度模型:调度器依次选择一个队列,队列中的一个作业,作业中的一个任务,最终将任务分配给有空闲slot的TaskTracker。assignTasks的实现也遵循这个模型:
Collection<JobInProgress> jobQueue = jobQueueJobInProgressListener.getJobQueue();对于FIFO调度器而言,队列即为对应监听器中使用的作业队列。然后,声明一个列表,用于保存选择的任务:
// Assigned tasks List<Task> assignedTasks = new ArrayList<Task>();接下来,计算队列中正在运行的和等待运行的map和reduce任务的数量:
// Compute (running + pending) map and reduce task numbers across pool int remainingReduceLoad = 0; int remainingMapLoad = 0; synchronized (jobQueue) { for (JobInProgress job : jobQueue) { if (job.getStatus().getRunState() == JobStatus.RUNNING) { remainingMapLoad += (job.desiredMaps() - job.finishedMaps()); if (job.scheduleReduces()) { remainingReduceLoad += (job.desiredReduces() - job.finishedReduces()); } } } }其中,job.scheduleReduces方法判断当前map任务的总体进度是否满足reduce任务开始调度的条件,map任务完成的比例是否超过变量mapred.reduce.slowstart.completed.maps的值,若超过则计算reduce任务的剩余任务数。接下来,计算map和reduce任务的负载因子:
// Compute the 'load factor' for maps and reduces double mapLoadFactor = 0.0; if (clusterMapCapacity > 0) { mapLoadFactor = (double)remainingMapLoad / clusterMapCapacity; } double reduceLoadFactor = 0.0; if (clusterReduceCapacity > 0) { reduceLoadFactor = (double)remainingReduceLoad / clusterReduceCapacity; }map任务负载因子定义为当前剩余的(正在执行的和等待开始的)map任务的总数与集群总的map资源数(map slot数目)的商值。reduce任务负载因子同理。计算负载因子的目的是根据TaskTracker的负载情况和集群总的负载情况将所有任务均衡地调度到各个TaskTracker以便均衡地使用各个结点上的资源。根据这种思想,可以计算出某个TaskTracker当前可用的slot数目:
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); }由此可见,可用slot定义为:根据集群总体负载均衡还有多少slot应该可用的数目减去实际已经在用的slot数目。注意,exceededMapPadding表示是否有足够的slot预留给推测执行的任务。所谓推测执行,是Hadoop为了防止某些任务执行过慢,为一些较慢任务启动一个备份任务,让该任务做相同的事情,并最终选用最先成功运行完成的任务计算结果为最终结果。推测执行机制日后关注。下面就是任务选择过程:
int numLocalMaps = 0; int numNonLocalMaps = 0; scheduleMaps: for (int i=0; i < availableMapSlots; ++i) { synchronized (jobQueue) { for (JobInProgress job : jobQueue) { if (job.getStatus().getRunState() != JobStatus.RUNNING) { continue; } Task t = null; // Try to schedule a node-local or rack-local Map task t = job.obtainNewNodeOrRackLocalMapTask(taskTrackerStatus, numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts()); if (t != null) { assignedTasks.add(t); ++numLocalMaps; // Don't assign map tasks to the hilt! // Leave some free slots in the cluster for future task-failures, // speculative tasks etc. beyond the highest priority job if (exceededMapPadding) { break scheduleMaps; } // Try all jobs again for the next Map task break; } // Try to schedule a node-local or rack-local Map task t = job.obtainNewNonLocalMapTask(taskTrackerStatus, numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts()); if (t != null) { assignedTasks.add(t); ++numNonLocalMaps; // We assign at most 1 off-switch or speculative task // This is to prevent TaskTrackers from stealing local-tasks // from other TaskTrackers. break scheduleMaps; } } } } int assignedMaps = assignedTasks.size();对于某个空闲的slot,从队列中选择一个正在执行的作业,并调用obtainNewNodeOrRackLocalMapTask方法获得一个具有数据本地性地任务。若找到了这样的任务,将其放入结果列表中,并检查刚才获得的exceedingMapPadding的值。若不满足,则跳出最外层循环,重新为每个slot分配任务,以期有新的空闲slot产生,从而满足推测执行的需求。当找到一个数据本地性任务后,马上跳出对队列的遍历,为下一个slot分配任务。
若没有找到具有数据本地性的任务,就调用obtainNewNonLocalMapTask方法获取一个非本地性的任务。如果找到了这样的任务,就将其放入结果列表中,然后跳出最外层循环,重新为每个slot分配任务。也就是说,一旦找到了一个非本地性任务,那么不能再继续获取任务,防止对于其他slot来说具有本地性地任务被抢夺。
这里解释一下数据本地性。在分布式环境中,为了减少任务执行过程中的网络传输开销,通常将任务调度到输入数据所在的计算节点,也就是让数据在本地进行计算 【1】 。Hadoop中数据本地性有三个等级:node-local(同节点),rack-local(同机架)和off-switch(跨机架)。选择任务时即按照上述顺序依次进行。
int target = findNewMapTask(tts, clusterSize, numUniqueHosts, maxLevel, status.mapProgress()); if (target == -1) { return null; } Task result = maps[target].getTaskToRun(tts.getTrackerName()); if (result != null) { addRunningTaskToTIP(maps[target], result.getTaskID(), tts, true); resetSchedulingOpportunities(); } return result;
其中,findNewMapTask方法的第四个参数指定了获取任务的本地性等级,maxLevel表示最高。在obtainNewNonLocalMapTask方法中则使用的是
NON_LOCAL_CACHE_LEVEL。在findNewMapTask方法中可以看到,运行失败的任务总是被优先选择,让它们能够快速重新执行;然后按照数据本地性选择尚未运行的任务;最后是查找正在运行的任务,为较慢的任务启动备份(推测执行)。有兴趣可以看源码这里不展示了。
对于reduce任务来说选择过程十分类似,只不过reduce任务不涉及数据本地性,因为它的输入来自map任务的输出,来自所有map任务的结点。
synchronized (jobQueue) { for (JobInProgress job : jobQueue) { if (job.getStatus().getRunState() != JobStatus.RUNNING || job.numReduceTasks == 0) { continue; } Task t = job.obtainNewReduceTask(taskTrackerStatus, numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts() ); if (t != null) { assignedTasks.add(t); break; } // Don't assign reduce tasks to the hilt! // Leave some free slots in the cluster for future task-failures, // speculative tasks etc. beyond the highest priority job if (exceededReducePadding) { break; } } }注意,每一次心跳只分配一个reduce任务。
最后,我们关注一下当要执行的任务获得以后,如何返回给TaskTracker,以及JobTracker下达的一些命令。
重新来看心跳方法heartbeat。它的返回值是一个HeartbeatResponse类型,其中有一个重要的字段:
TaskTrackerAction[] actions;这个数组就用于JobTracker向TaskTracker下达命令,包括执行刚刚选择的任务的指令。具体的命令种类有以下五种:
1. ReinitTrackerAction
2. LaunchTaskAction
3. KillTaskAction
4. KillJobAction
5. CommitTaskAction
两种情况下JobTracker会下达ReinitTrackerAction命令:丢失上次心跳应答信息或者丢失TaskTracker状态信息。这两种状态为不一致状态。
short newResponseId = (short)(responseId + 1); status.setLastSeen(now); if (!processHeartbeat(status, initialContact, now)) { if (prevHeartbeatResponse != null) { trackerToHeartbeatResponseMap.remove(trackerName); } return new HeartbeatResponse(newResponseId, new TaskTrackerAction[] {new ReinitTrackerAction()}); }
LaunchTaskAction命令即包含了需要执行的任务。JobTracker在选择任务时首先选择的是辅助型任务,例如job-cleanup task,task-cleanup task和job-setup task。这些任务在调用assignTasks方法之前就已经选择,因此优先级最高。
List<Task> tasks = getSetupAndCleanupTasks(taskTrackerStatus); if (tasks == null ) { tasks = taskScheduler.assignTasks(taskTrackers.get(trackerName)); } if (tasks != null) { for (Task task : tasks) { expireLaunchingTasks.addNewTask(task.getTaskID()); actions.add(new LaunchTaskAction(task)); } }
KillTaskAction封装了需要杀死的任务。杀死的原因可能是任务失败,用户通过kill命令杀死等。KillJobAction封装了待清理的作业。清理的工作主要是删除临时目录。作业完成或失败时都会导致该作业被清理。最后,CommitTaskAction封装了需要提交的任务。Hadoop将一个成功运行完成的Task Attempt(一个任务的多个备份任务)结果文件从临时目录转移到最终目录的过程称为任务提交。后三种命令生成的代码如下:
// Check for tasks to be killed List<TaskTrackerAction> killTasksList = getTasksToKill(trackerName); if (killTasksList != null) { actions.addAll(killTasksList); } // Check for jobs to be killed/cleanedup List<TaskTrackerAction> killJobsList = getJobsForCleanup(trackerName); if (killJobsList != null) { actions.addAll(killJobsList); } // Check for tasks whose outputs can be saved List<TaskTrackerAction> commitTasksList = getTasksToSave(status); if (commitTasksList != null) { actions.addAll(commitTasksList); }至此,任务调度功流程大体框架全部结束,接下来就是任务在TaskTracker上的具体执行过程了。请关注后续文章。