【hadoop代码笔记】hadoop作业提交之TaskTracker 启动task

一、概要描述

上篇博文描 述了TaskTracker从Jobtracker如何从JobTracker获取到要执行的Task。在从JobTracker获取到 LaunchTaskAction后,执行addToTaskQueue方法来把要执行的Task加入到queue。在本篇博文中,我们来关注下该方法 后,TaskTracker怎么来处理这些Task。

实际上,TaskTracker初始化时,会初始化并启动两个TaskLauncher类型的线程,mapLauncher,reduceLauncher。在TaskTracker从JobTracher获取到任务后,对应的会把任务添加到两个 TaskLauncher的Queue中,其实是TaskLauncher维护的一个列表List<TaskInProgress> tasksToLaunch。
TaskLauncher线程一直会定时检查TaskTracher上面有slot开业运行新的Task,则启动 Task。在这个过程中,先把task运行需要的文件解压到本地,并创建根据Task类型(Map或者Reduce)创建一个TaskRunner线程, 在TaskRunner中JvmManager调用JvmManagerForType、JvmRunner来启动一个java进程来执行Map或Reduce任务。

本文只是介绍到启动一个java进程,至于是什么样的java进程,对于maptask和reducetask分别是怎么执行的,在后面的child启动maptask,和child启动reducetask 会比较详细的介绍。

二、 流程描述

1.tasktracker的offerService方法获取到要执行的task后调用addToTaskQueue方法,其实是调用taskrunner的addToTaskQueue方法
2.TaskLauncher内部维护了一个List<TaskInProgress> tasksToLaunch,只是把task加入到该集合中
3.taskLauncher是一个线程,在其run方法中从tasksToLaunch集合中取出task来执行,调用Tasktracker的startNewTask方法启动task。
4. startNewtask方法中调用localizeJob方法把job相关的配置信息和要运行的jar拷贝到tasktracker本地,然后调用taskInProgress的launchTask方法来启动task。
5.TaskInProgress的launchTask方法先调用localizeTask(task把task相关的配置信息获取到本地。然后创建一个TaskRunner线程来启动task。
6.在TaskRunner的run方法中构建一个java命令的执行的条件,包括引用类,执行目录等,入口类是Child。然后调用JvmManager 的launchJvm方法来调用。
7.JvmManager 进而调用 JvmManagerForType的reapJvm,和spawnNewJvm 方法,发起调用。
8. 在JvmManagerForType的spawnNewJvm 方法中创建了一个JvmRunner线程类执行调用。
9. JvmRunner线程的run反复调用runChild方法来执行 一个命令行的调用。

hadoop_mapreduce_tasktracker_launch_task

TaskTracker 启动task

三、代码详细

1. TaskTracker的 addToTaskQueue方法。

接上文的最后一个方法的在heartbeat中把根据jobtracker的指令把需要launch的task调用addToTaskQueue方法加入task queue。

2. TaskLauncher 的addToTaskQueue方法,即把要launch的task加入到TaskLauncher内维护的一个列表List<TaskInProgress> tasksToLaunch;中。

3. TaskLauncher线程的run方法。TaskLauncher是一个线程。一直检查task列表中有数据,取出一个来执行。

4. TaskTracker的startNewTask 启动一个新task。该方法的主要代码就一句。

5. TaskTracker的localizeJob方法。 初始化job的目录

6. TaskTracker的addTaskToJob方法。只是把job和task的关系加入到runningJobs中。

7. TaskTracker的launchTaskForJob方法。调用TaskInprogress的launchTask方法。

8. TaskIProgress的 launchTask方法。

9.TaskinProgress的localizeTask方法。把Task相关的文件拷贝到本地。

10.Task是个抽象类,两个子类分别是MapTask和ReduceTask。先关注Map的TaskRunner。

11. TaskRunner线程的Run方法,有420行代码!主要作用是根据配置信息,构造java命令,启动一个java进程。

拼接一个java指令,启动一个单独的java进程来执行每一个map或者reduce任务。这个java命令的class是Child。即这个java进程最终调用的是Child类的main函数。

TaskRunner的run方法

12.JvmManager的 launchJvm方法。在TaskRunner的run方法,是构造一个java命令的参数,调用JvmManager的launchJvm方法执行。

13. JvmManagerForType的reapJvm方法

14. JvmManagerForType的spawnNewJvm方法。重新启动一个jvm。

15. JvmRunner线程的run方法。

16. JvmRunner线程的runChild方法。其中掉ShellCommandExecutor的execute方法。 ShellCommandExecutor封装了shell执行。即把前面步骤构造的JvmEnv类型的执行信息分装成一个字符串列表,使用该列表构造一 个ShellCommandExecutor来执行命令。

完.

原创文章。为了维护文章的版本一致、最新、可追溯,转载请注明: 转载自idouba

本文链接地址: 【hadoop代码笔记】hadoop作业提交之TaskTracker 启动task


, , ,

No comments yet.

发表评论