我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。
TaskTracker周期性地向JobTracker发送心跳报告,在RPC调用返回结果后,解析结果得到JobTracker下发的运行Task的指令,即LaunchTaskAction,就会在TaskTracker节点上准备运行这个Task。Task的运行是在一个与TaskTracker进程隔离的JVM实例中执行,该JVM实例是通过org.apache.hadoop.mapred.Child来创建的,所以在创建Child VM实例之前,需要做大量的准备工作来启动Task运行。一个Task的启动过程,如下序列图所示:
通过上图,结合源码,我们将一个Task启动的过程,分为下面3个主要的步骤:
下面,我们详细分析上面3个步骤的流程:
如果是LaunchTaskAction,则TaskTracker会将该指令加入到一个启动Task的队列中,进行一步加载处理,如下所示:
private void addToTaskQueue(LaunchTaskAction action) { if (action.getTask().isMapTask()) { mapLauncher.addToTaskQueue(action); } else { reduceLauncher.addToTaskQueue(action); } }
根据Task的类型,分别加入到对应类型的TaskLauncher的队列中。这里需要了解一下TaskLauncher线程类,在TaskTracker中创建了2个TaskLauncher线程,一个是为启动MapTask,另一个是为启动ReduceTask。下面是TaskLauncher类的构造方法:
public TaskLauncher(TaskType taskType, int numSlots) { this.maxSlots = numSlots; this.numFreeSlots = new IntWritable(numSlots); this.tasksToLaunch = new LinkedList<TaskInProgress>(); setDaemon(true); setName("TaskLauncher for " + taskType + " tasks"); }
构造方法中,参数taskType表示Task类型,分为MapTask和ReduceTask,参数numSlots表示对每一种类型的Task每个TaskTracker上最多可以启动的Task的实例数,默认都是2个。在TaskTracker初始化时,会读取mapred-site.xml配置文件,读取mapred.tasktracker.map.tasks.maximum和mapred.tasktracker.reduce.tasks.maximum配置的参数值,分别赋值给maxMapSlots和maxReduceSlots这2个属性,如下TaskTracker构造方法中初始化这2个属性:
maxMapSlots = conf.getInt("mapred.tasktracker.map.tasks.maximum", 2); maxReduceSlots = conf.getInt("mapred.tasktracker.reduce.tasks.maximum", 2);
然后,在TaskTracker创建时,会根据上述maxMapSlots和maxReduceSlots的值来创建并启动2个TaskLauncher线程:
mapLauncher = new TaskLauncher(TaskType.MAP, maxMapSlots); reduceLauncher = new TaskLauncher(TaskType.REDUCE, maxReduceSlots); mapLauncher.start(); reduceLauncher.start();
将LaunchTaskAction加入到TaskLauncher的队列中,这个是调用TaskLauncher的addToTaskQueue()方法:
public void addToTaskQueue(LaunchTaskAction action) { synchronized (tasksToLaunch) { TaskInProgress tip = registerTask(action, this); // 注册Task,初始化用来跟踪该待启动的Task相关的数据结构 tasksToLaunch.add(tip); // 将TIP加入队列 tasksToLaunch.notifyAll(); // 通知TaskLauncher线程自己(在run()方法中会调用wait())启动Task } }
上面方法中,最关键的就是registerTask()方法,调用该方法来初始化TaskTracker端Task对应TaskInProgress结构,代码如下所示:
private TaskInProgress registerTask(LaunchTaskAction action, TaskLauncher launcher) { Task t = action.getTask(); LOG.info("LaunchTaskAction (registerTask): " + t.getTaskID() + " task's state:" + t.getState()); TaskInProgress tip = new TaskInProgress(t, this.fConf, launcher); // 创建TIP synchronized (this) { tasks.put(t.getTaskID(), tip); // 加入到队列tasks:TaskAttemptID -> TaskInProgress runningTasks.put(t.getTaskID(), tip); // 加入到队列runningTasks:TaskAttemptID -> TaskInProgress boolean isMap = t.isMapTask(); if (isMap) { mapTotal++; } else { reduceTotal++; } } return tip; }
上面方法中,tasks队列用来记录该TaskTracker上所有的Task,包括正在运行和已经完成的Task,而队列runningTasks则表示当前TaskTracker上正在运行的Task。同时,通过mapTotal和reduceTotal来分别记录当前TaskTracker上运行的总的MapTask和ReduceTask的数量。根据LaunchTaskAction创建的TaskInProgress结构被加入到队列tasksToLaunch中,然后通知TaskLauncher线程,在方法run中检测并取出队列中TaskInProgress对象,并判断当前TaskTracker的资源状态能否启动一个Task,如果可以则调用startNewTask()方法启动Task,代码如下所示:
TaskInProgress tip; Task task; synchronized (tasksToLaunch) { while (tasksToLaunch.isEmpty()) { // 队列为空,则没有Task需要启动,等待向队列加入LaunchTaskAction指令及其通知 tasksToLaunch.wait(); } tip = tasksToLaunch.remove(0); // 队列不空,则取出LaunchTaskAction task = tip.getTask(); LOG.info("Trying to launch : " + tip.getTask().getTaskID() + " which needs " + task.getNumSlotsRequired() + " slots"); } synchronized (numFreeSlots) { // 检查当前是否存在空闲的slot,以便运行Task boolean canLaunch = true; while (numFreeSlots.get() < task.getNumSlotsRequired()) { // 如果当前空闲slot小于该Task运行所需的slot数量 if (!tip.canBeLaunched()) { // 如果TIP状态不是下面3种状态:UNASSIGNED、FAILED_UNCLEAN、KILLED_UNCLEAN canLaunch = false; // 检查TIP状态不能启动Task,但也不能阻塞该方法 break; } LOG.info("TaskLauncher : Waiting for " + task.getNumSlotsRequired() + " to launch " + task.getTaskID() + ", currently we have " + numFreeSlots.get() + " free slots"); numFreeSlots.wait(); // 如果没有空闲slot,则等待 } if (!canLaunch) { continue; } LOG.info("In TaskLauncher, current free slots : " + numFreeSlots.get()+ " and trying to launch "+tip.getTask().getTaskID() + " which needs " + task.getNumSlotsRequired() + " slots"); numFreeSlots.set(numFreeSlots.get() - task.getNumSlotsRequired()); // 标记将满足该Task的Slot数已经分配 assert (numFreeSlots.get() >= 0); } synchronized (tip) { // 到这里已经获取到了满足运行Task要求的空闲slot,但还要检查该TIP状态是否指示为被kill了 if (!tip.canBeLaunched()) { LOG.info("Not launching task " + task.getTaskID() + " as it got killed externally. Task's state is " + tip.getRunState()); addFreeSlots(task.getNumSlotsRequired()); // 如果Task状态TIP标识不能启动,则释放slot continue; } tip.slotTaken = true; } startNewTask(tip); // 获取到了满足Task启动所需的空闲slot,开始启动Task
这样,当前TaskTracker所在节点的资源状态,和Task对应的TIP状态都已经满足启动Task的要求,可以启动一个Task去运行。
调用startNewTask()方法,异步地启动了一个单独的线程去启动Task,该方法如下所示:
void startNewTask(final TaskInProgress tip) throws InterruptedException { Thread launchThread = new Thread(new Runnable() { @Override public void run() { try { RunningJob rjob = localizeJob(tip); // 在TaskTracker节点上初始化Job信息 tip.getTask().setJobFile(rjob.getLocalizedJobConf().toString()); launchTaskForJob(tip, new JobConf(rjob.getJobConf()), rjob); // 启动Task } catch (Throwable e) { ... ... } } }); launchThread.start(); }
如果在一个TaskTracker节点上运行的多个Task都属于同一个Job(一个TaskTracker上运行的Task按照Job来分组,每一组Task都属于同一个Job),那么第一次初始化时,还没有建立一个Task到Job的映射关系,也就是说,在TaskTracker端也要维护Job的状态,以及属于该Job的所有Task的状态信息。比如,如果用户提交了一个kill掉Job的请求,那么正在运行的属于该Job的所有Task都应该被kill掉。上面代码中调用localizeJob()方法,执行了如下处理:
这里,TaskController使用的LinuxTaskController实现类,通过调用该方法,实际上构造了一个Shell命令行,用来在TaskTracker节点上初始化目录和拷贝相关资源,该命令行示例如下所示:
/usr/local/java/bin/java -classpath .:/usr/local/java/lib/*.jar;/usr/local/java/jre/lib/*.jar -Dhadoop.log.dir=/tmp/hadoop/logs -Dhadoop.root.logger=INFO,console -Djava.library.path= org.apache.hadoop.mapred.JobLocalizer shirdrn job_200912121733_0002
通过工具ShellCommandExecutor来执行上述命令行,启动一个单独的JVM实例,完成Job资源初始化,完成即退出。通过上述命令行可以看到,主要的初始化工作都在JobLocalizer中完成的,需要传入2个参数:用户、jobid,在JobLocalizer中创建了一个Job所包含的各种资源,供Task在TaskTracker节点上运行共享,这些相关的目录或资源文件包括:
${mapred.local.dir}/taskTracker/${user} ${mapred.local.dir}/taskTracker/${user}/jobcache ${mapred.local.dir}/taskTracker/${user}/jobcache/${jobid}/work ${mapred.local.dir}/taskTracker/${user}/jobcache/${jobid}/jars ${mapred.local.dir}/taskTracker/${user}/jobcache/${jobid}/jars/job.jar ${mapred.local.dir}/taskTracker/${user}/jobcache/${jobid}/job.xml ${mapred.local.dir}/taskTracker/${user}/jobcache/${jobid}/jobToken ${mapred.local.dir}/taskTracker/${user}/distcache
这样,在一个TaskTracker节点上运行的一组Task所共享的对应唯一Job相关的资源已经满足,接下来就可以启动Task了。
启动Task的流程相对复杂一些,我们分几个阶段/要点来进行说明:
在startNewTask()方法中调用localizeJob()方法,完成了Job资源在TaskTracker节点上的初始化,接着就可以调用launchTaskForJob()方法进入启动Task的处理流程,代码如下所示:
protected void launchTaskForJob(TaskInProgress tip, JobConf jobConf, RunningJob rjob) throws IOException { synchronized (tip) { jobConf.set(JobConf.MAPRED_LOCAL_DIR_PROPERTY, localStorage.getDirsString()); tip.setJobConf(jobConf); tip.setUGI(rjob.ugi); tip.launchTask(rjob); // 这里才是启动Task的核心方法 } }
通过调用TaskInProgress tip的launchTask()方法来启动Task,我们看一下该方法实现代码:
public synchronized void launchTask(RunningJob rjob) throws IOException { if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED || this.taskStatus.getRunState() == TaskStatus.State.FAILED_UNCLEAN || this.taskStatus.getRunState() == TaskStatus.State.KILLED_UNCLEAN) { localizeTask(task); if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED) { // 如果状态UNASSIGNED,则初始化完成后,将当前状态改为RUNNING this.taskStatus.setRunState(TaskStatus.State.RUNNING); } setTaskRunner(task.createRunner(TaskTracker.this, this, rjob)); // 启动一个TaskRunner线程 this.runner.start(); long now = System.currentTimeMillis(); this.taskStatus.setStartTime(now); this.lastProgressReport = now; } else { LOG.info("Not launching task: " + task.getTaskID() + " since it's state is " + this.taskStatus.getRunState()); } }
TaskInProgress里面taskStatus维护了一个TIP的状态,通过上述代码可以看出,一个Task只有具备下面3个状态之一:UNASSIGNED、FAILED_UNCLEAN、KILLED_UNCLEAN,才能够被启动:首先要进行Task的初始化,调用localizeTask()方法,如下所示:
void localizeTask(Task task) throws IOException{ task.localizeConfiguration(localJobConf); task.setConf(localJobConf); }
在这里,Task可能是MapTask,也可能是ReduceTask,所以调用task.localizeConfiguration()的初始化逻辑稍微有些不同,具体可以查看MapTask和ReduceTask类实现。另外,对于不同类型的Task,也会创建不同类型的TaskRunner线程,分别对应于MapTaskRunner和ReduceTaskRunner,实际所有Task启动的相关逻辑都是在这2个TaskRunner中实现的。在TaskRunner中,主要逻辑是在run()方法中实现的,其中在调用launchJvmAndWait(setupCmds, vargs, stdout, stderr, logSize, workDir)之前,做了一些准备工作:
完成上述准备工作以后,调用launchJvmAndWait()方法,创建Child VM实例,如下所示:
void launchJvmAndWait(List <String> setup, Vector<String> vargs, File stdout, File stderr, long logSize, File workDir) throws InterruptedException, IOException { jvmManager.launchJvm(this, jvmManager.constructJvmEnv(setup, vargs, stdout, stderr, logSize, workDir, conf)); synchronized (lock) { while (!done) { lock.wait(); } } }
最终是通过JvmManager来实现JVM实例的创建,下面是JvmManager保存的一些数据结构,用来维护JVM相关数据的数据结构,如下图所示:
可以看到,一个JvmManager对应2个JvmManagerForType,分别负责管理MapTask和ReduceTask启动对应的Child VM等数据,JvmManager的构造方法,如下所示:
public JvmManager(TaskTracker tracker) { mapJvmManager = new JvmManagerForType(tracker.getMaxCurrentMapTasks(), true, tracker); reduceJvmManager = new JvmManagerForType(tracker.getMaxCurrentReduceTasks(), false, tracker); }
上面调用了jvmManager.launchJvm()方法,其中内部根据Task类型,选择调用mapJvmManager或reduceJvmManager的reapJvm()方法,如下所示:
private synchronized void reapJvm(TaskRunner t, JvmEnv env) throws IOException, InterruptedException { if (t.getTaskInProgress().wasKilled()) { // 检查当前准备启动的Task是否已经被kill掉,如果是则直接返回 return; } boolean spawnNewJvm = false; JobID jobId = t.getTask().getJobID(); int numJvmsSpawned = jvmIdToRunner.size(); JvmRunner runnerToKill = null; // 检查是否存在空闲slot来启动Task if (numJvmsSpawned >= maxJvms) { // 检查当前TaskTracker上运行的某类Task对应的JVM实例数是否大于全局设置允许的最大slot数 Iterator<Map.Entry<JVMId, JvmRunner>> jvmIter = jvmIdToRunner.entrySet().iterator(); while (jvmIter.hasNext()) { // 遍历当前存在的<JVMId, JvmRunner>队列,检查每个JvmRunner的状态 JvmRunner jvmRunner = jvmIter.next().getValue(); JobID jId = jvmRunner.jvmId.getJobId(); if (jId.equals(jobId) && !jvmRunner.isBusy() && !jvmRunner.ranAll()){ // 如果在当前要启动的Task之前,已经有该Task对应的Job的其他Task运行完成,则预留该JVM以重用 setRunningTaskForJvm(jvmRunner.jvmId, t); // 将该JvmRunner映射到当前Task LOG.info("No new JVM spawned for jobId/taskid: " + jobId+"/"+t.getTask().getTaskID() + ". Attempting to reuse: " + jvmRunner.jvmId); return; } //一个JVM实例需要kill掉,需要满足下面条件之一: // (1) 如果属于当前要启动的Task对应的Job,该Job对应的其他Task都已经运行完成; // (2) 如果不属于当前要启动的Task所对应的Job,那些Job对应的Task都已经运行完成。 if ((jId.equals(jobId) && jvmRunner.ranAll()) || (!jId.equals(jobId) && !jvmRunner.isBusy())) { runnerToKill = jvmRunner; spawnNewJvm = true; } } } else { spawnNewJvm = true; } if (spawnNewJvm) { if (runnerToKill != null) { LOG.info("Killing JVM: " + runnerToKill.jvmId); killJvmRunner(runnerToKill); // kill掉该JvmRunner } spawnNewJvm(jobId, env, t); // 创建一个新的JVM实例来启动该Task return; } } catch (Exception e) { LOG.fatal(e); } finally { System.exit(-1); } }
上面代码中,调用setRunningTaskForJvm()很关键,实际上把需要启动的Task与JvmRunner建立映射关系,更新相应的内存数据结构(队列),如下所示:
synchronized public void setRunningTaskForJvm(JVMId jvmId, TaskRunner t) { jvmToRunningTask.put(jvmId, t); runningTaskToJvm.put(t,jvmId); jvmIdToRunner.get(jvmId).setBusy(true); // 设置当前JvmRunner被占用,不允许释放该资源 }
该方法,在spawnNewJvm()方法也调用了,spawnNewJvm()方法创建了一个新的JVM,代码如下所示:
private void spawnNewJvm(JobID jobId, JvmEnv env, TaskRunner t) { JvmRunner jvmRunner = new JvmRunner(env, jobId, t.getTask()); jvmIdToRunner.put(jvmRunner.jvmId, jvmRunner); jvmRunner.setDaemon(true); jvmRunner.setName("JVM Runner " + jvmRunner.jvmId + " spawned."); setRunningTaskForJvm(jvmRunner.jvmId, t); // 调用setRunningTaskForJvm()方法 LOG.info(jvmRunner.getName()); jvmRunner.start(); // 启动JvmRunner线程,用来启动Child VM }
接下来,我们看一下JvmRunner线程类,该线程体run()方法中直接调用了runChild()方法,该方法实现代码,如下所示:
public void runChild(JvmEnv env) throws IOException, InterruptedException{ int exitCode = 0; try { env.vargs.add(Integer.toString(jvmId.getId())); TaskRunner runner = jvmToRunningTask.get(jvmId); // 从队列jvmToRunningTask中取出TaskRunner if (runner != null) { Task task = runner.getTask(); String user = task.getUser(); TaskAttemptID taskAttemptId = task.getTaskID(); String taskAttemptIdStr = task.isTaskCleanupTask() ? (taskAttemptId.toString() + TaskTracker.TASK_CLEANUP_SUFFIX) : taskAttemptId.toString(); exitCode = tracker.getTaskController().launchTask(user, jvmId.jobId.toString(), taskAttemptIdStr, env.setup, env.vargs, env.workDir, env.stdout.toString(), env.stderr.toString()); // 通过TaskController来启动Task } } catch (IOException ioe) { // do nothing } finally { // handle the exit code // although the process has exited before we get here, // make sure the entire process group has also been killed. kill(); // Task运行完成,kill掉运行Task的Child VM实例 updateOnJvmExit(jvmId, exitCode); LOG.info("JVM : " + jvmId + " exited with exit code " + exitCode + ". Number of tasks it ran: " + numTasksRan); deleteWorkDir(tracker, firstTask); // 清理临时目录 } }
在JvmRunner线程类中,其中委托TaskController来控制Task的实际启动。
下面,我们看TaskController启动Task的实现方法launchTask(),代码如下所示:
@Override public int launchTask(String user, String jobId, String attemptId, List<String> setup, List<String> jvmArguments, File currentWorkDirectory, String stdout, String stderr) throws IOException { ShellCommandExecutor shExec = null; try { FileSystem rawFs = FileSystem.getLocal(getConf()).getRaw(); long logSize = 0; //TODO MAPREDUCE-1100 String cmdLine = TaskLog.buildCommandLine(setup, jvmArguments, new File(stdout), new File(stderr), logSize, true); // 将上述命令行写入本地缓存的目录中 Path p = new Path(allocator.getLocalPathForWrite( TaskTracker.getPrivateDirTaskScriptLocation(user, jobId, attemptId), getConf()), COMMAND_FILE); String commandFile = writeCommand(cmdLine, rawFs, p); String[] command = new String[]{taskControllerExe, user, localStorage.getDirsString(), Integer.toString(Commands.LAUNCH_TASK_JVM.getValue()), jobId, attemptId, currentWorkDirectory.toString(), commandFile}; shExec = new ShellCommandExecutor(command); if (LOG.isDebugEnabled()) { LOG.debug("launchTask: " + Arrays.toString(command)); } shExec.execute(); // 执行启动Task的命令 } catch (Exception e) { if (shExec == null) { return -1; } int exitCode = shExec.getExitCode(); LOG.warn("Exit code from task is : " + exitCode); // 143 (SIGTERM) and 137 (SIGKILL) exit codes means the task was // terminated/killed forcefully. In all other cases, log the // task-controller output if (exitCode != 143 && exitCode != 137) { LOG.warn("Exception thrown while launching task JVM : " + StringUtils.stringifyException(e)); LOG.info("Output from LinuxTaskController's launchTaskJVM follows:"); logOutput(shExec.getOutput()); } return exitCode; } if (LOG.isDebugEnabled()) { LOG.debug("Output from LinuxTaskController's launchTask follows:"); logOutput(shExec.getOutput()); } return 0; }
将构造好的启动Child的命令行写入到本地目录下的文件中,该脚本文件的绝对路径,示例如下所示:
/tmp/mapred/local/ttprivate/taskTracker/shirdrn/jobcache/job_200912121733_0002/attempt_200912121733_0002_m_000005_0/taskjvm.sh
在TaskController(实际上是LinuxTaskController)的launchTask()方法中,使用ShellCommandExecutor工具执行的命令行,类似如下这样:
/usr/local/hadoop/bin/task-controller shirdrn /tmp/mapred/local 1 job_200912121733_0002 attempt_200912121733_0002_m_000005_0 /tmp/mapred/local/ttprivate/taskTracker/shirdrn/jobcache/job_200912121733_0002/attempt_200912121733_0002_m_000005_0/taskjvm.sh
在taskjvm.sh脚本中的内容,才是真正启动Child VM的命令行,示例如下所示:
/usr/local/bin/java -Xmx 512M -verbose:gc -Xloggc:/tmp/attempt_200912121733_0002_m_000005_0.gc -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.library.path= -Djava.io.tmpdir= -classpath .:/usr/local/java/lib/*.jar:/usr/local/java/jre/lib/*.jar -Dlog4j.configuration=task-log4j.properties -Dhadoop.log.dir=/tmp/hadoop/logs -Dhadoop.root.logger=INFO,TLA -Dhadoop.tasklog.taskid=attempt_200912121733_0002_m_000005_0 -Dhadoop.tasklog.iscleanup=false -Dhadoop.tasklog.totalLogFileSize= org.apache.hadoop.mapred.Child 127.0.0.1 0 attempt_200912121733_0002_m_000005_0 /tmp/hadoop/logs/userlogs/job_200912121733_0002/attempt_200912121733_0002_m_000005_0/ 2134
至此,一个Task通过Child VM的加载已经启动,就可以运行一个Task了,我们后续再详细介绍。
本文基于 署名-非商业性使用-相同方式共享 4.0 许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系。