Task在TaskTracker上执行环境的准备

深藏阁楼爱情的钟 2022-06-17 11:06 242阅读 0赞

在前面的文章中，详细的介绍了Job的任务分解以及Job任务的调度，同时也讲述了TaskTracker是每一通过向JobTracker发送心跳包heartbeat并从其响应中获取自己被分配的任务的。那么，在TaskTracker中，它是如何来执行这些Task的呢，或者说为这个Task的执行做了哪些准备工作呢？

其实，在一个TaskTracker节点中，每一Task的执行并不是在当前这个TaskTracker所属的进程中完成的，而是对每一个Task，都从新开启一个对应的JVM进程，让这个JVM进程来执行这个Task，同时这个JVM进程通过网络通行的方式来和开启它的TaskTracker进行交互，报告对应的Task的执行进度。这个执行模型可简单理解为：

![0_1323677956U2c2.gif][]

在这里，我将详细为大家讨论TaskTracker是如何从接收到Task，到开始执行这个Task中间的过程，至于Child是如何执行Task的，在本文我不会作详细的讨论(但我会在以后的博文中详细的阐述)。我把这个过程成为Task的执行准备阶段，先来看看与之相关的类图吧！这个类体系图相当的复杂：

![0_1323681988zNe3.gif][]

好了，再来看看与这些相关的时序图，以便更详细地了解TaskTracker是如何为一个新来的Task准备JVM运行环境的。

![0_1323690022Rkm9.gif][]

了解了每一个先关类及其方法的调用时序关系之后，就来看看这些类中的一些重要方法到底都具体干了那些事？

1.TaskTracker.registerTask()

该方法为新来的一个Task创建一个对应的TaskInProgress对象，同时在任务集合tasks和正在运行的任务集合中runningTasks中保存Task和TaskInProgress的映射关系，并根据Task的类型更新当前TaskTracker的map/reduce任务数量。

2.TaskTracker.addTaskToJob()

检查该Task所属的Job是否已在TaskTracker运行，如果没有，则创建该Job的一个RunningJob对象，并把该Task放到RunningJob中，最后放回该Task‘s Job的RunningJob。

3.TaskTracker.localizeJob()

如果Task对应的RunningJob还没有本地化，则需要本地化该Job。本地化该Job实际上就是根据JobID信息从HDFS中下载该Job对应的配置文件job.xml和运行文件job.jar到本地的文件系统。TaskTracker会为每一个Job创建一个目录，这个目录放在$(mapred.local.dir)/taskTracker/jobcache/下面，这里的$(mapred.local.dir)指的是mapreduce的配置文件中的mapred.local.dir项的值。Job的job.xml文件被放在Job的本地目录下：$(mapred.local.dir)/taskTracker/jobcache/jobid/job.xml，而Job的运行文件job.jar被放在Job的本地目录的子目录jars/下：$(mapred.local.dir)/taskTracker/jobcache/jobid/jars/job.jar，然后根据当前Job的实际配置信息重写job.xml文件，同时解压文件job.jar。

4.TaskInProgress.localizeTask()

首先会为该Task创建一个本地目录，目录名是该Task的id，这个目录在Task所属的Job的本地目录下：$(mapred.local.dir)/taskTracker/jobcache/jobid/taskid/，如果该Task的本地目录已存在，则需先清空该Task的本地目录。同时，在该目录下创建一个该Task被调用执行JVM时的工作目录：$(mapred.local.dir)/taskTracker/jobcache/jobid/taskid/work/，然后将当前Task的详细配置信息写入Task的配置文件$(mapred.local.dir)/taskTracker/jobcache/jobid/taskid/job.xml中。

5.TaskRunner.run()

TaskRunner会将Task运行所依赖的第三方工具包、归档文件、普通文件等从HDFS复制到TaskTracker的本地文件系统，并且包这些文件缓存起来，以供其他的Task使用。这些文件被保存在$(mapred.local.dir)/taskTracker/archive/下面，同时这些文件的相对路径在$(mapred.local.dir)/taskTracker/archive/下保持不变。然后会为启动该Task对应的JVM进程配置运行环境和参数。最后会通过Java程序来调用系统命令，即启动一个JVM进程来执行该Task，关于如何使用Java程序来调用系统命令请参看我转载的系列文章：[Java调用系统命令学习][Java]。最后这个调用交给了ShellCommandExecutor来运行，还是举个例子来说明这个调用形式：

bash -c "echo $$ > /home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/attempt_201112121911_0001_m_000003_0/pid ;exec'/opt/jdk1.6.0/jre/bin/java' '-Djava.library.path=/opt/jdk1.6.0/jre/lib/i386/server:/opt/jdk1.6.0/jre/lib/i386:/opt/jdk1.6.0/jre/../lib/i386:/opt/jdk1.6.0/jre/lib/i386/client:/opt/jdk1.6.0/jre/lib/i386::/usr/java/packages/lib/i386:/lib:/usr/lib:/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/attempt_201112121911_0001_m_000003_0/work' '-Xmx200m' '-Djava.io.tmpdir=/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/attempt_201112121911_0001_m_000003_0/work/tmp' '-classpath''/home/xhh/workspace/Java/HMapred/bin:/home/xhh/workspace/Java/HMapred/lib/ant.jar:/home/xhh/workspace/Java/HMapred/lib/commons-cli-1.2.jar:/home/xhh/workspace/Java/HMapred/lib/commons-codec-1.3.jar:/home/xhh/workspace/Java/HMapred/lib/commons-el-1.0.jar:/home/xhh/workspace/Java/HMapred/lib/commons-httpclient-3.0.1.jar:/home/xhh/workspace/Java/HMapred/lib/commons-logging-1.0.4.jar:/home/xhh/workspace/Java/HMapred/lib/commons-logging-api-1.0.4.jar:/home/xhh/workspace/Java/HMapred/lib/commons-net-1.4.1.jar:/home/xhh/workspace/Java/HMapred/lib/core-3.1.1.jar:/home/xhh/workspace/Java/HMapred/lib/hadoop-0.20.2-tools.jar:/home/xhh/workspace/Java/HMapred/lib/hsqldb-1.8.0.10.jar:/home/xhh/workspace/Java/HMapred/lib/hsqldb-1.8.0.10.LICENSE.txt:/home/xhh/workspace/Java/HMapred/lib/jasper-compiler-5.5.12.jar:/home/xhh/workspace/Java/HMapred/lib/jasper-runtime-5.5.12.jar:/home/xhh/workspace/Java/HMapred/lib/jets3t-0.6.1.jar:/home/xhh/workspace/Java/HMapred/lib/jetty-6.1.14.jar:/home/xhh/workspace/Java/HMapred/lib/jetty-util-6.1.14.jar:/home/xhh/workspace/Java/HMapred/lib/jsp-2.1.jar:/home/xhh/workspace/Java/HMapred/lib/jsp-api-2.1.jar:/home/xhh/workspace/Java/HMapred/lib/junit-3.8.1.jar:/home/xhh/workspace/Java/HMapred/lib/kfs-0.2.2.jar:/home/xhh/workspace/Java/HMapred/lib/kfs-0.2.LICENSE.txt:/home/xhh/workspace/Java/HMapred/lib/log4j-1.2.15.jar:/home/xhh/workspace/Java/HMapred/lib/mockito-all-1.8.0.jar:/home/xhh/workspace/Java/HMapred/lib/oro-2.0.8.jar:/home/xhh/workspace/Java/HMapred/lib/servlet-api-2.5-6.1.14.jar:/home/xhh/workspace/Java/HMapred/lib/slf4j-api-1.4.3.jar:/home/xhh/workspace/Java/HMapred/lib/slf4j-log4j12-1.4.3.jar:/home/xhh/workspace/Java/HMapred/lib/xmlenc-0.52.jar:/home/xhh/workspace/Java/HMapred/lib/hdfs.jar::/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/jars/classes:/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/jars:/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/archive/mapred-fs/home/xhh/workspace/Java/HMapred/build/test/mapred/job/system/job_201112121911_0001/libjars/ant1.jar:/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/archive/mapred-fs/home/xhh/workspace/Java/HMapred/build/test/mapred/job/system/job_201112121911_0001/libjars/ant.jar:/home/xhh/workspace/Java/HMapred/build/test/mapred/task/local/taskTracker/jobcache/job_201112121911_0001/attempt_201112121911_0001_m_000003_0/work' '-Dhadoop.log.dir=/home/xhh/workspace/Java/HMapred/build/test/mapred/task/log' '-Dhadoop.root.logger=INFO,TLA' '-Dhadoop.tasklog.taskid=attempt_201112121911_0001_m_000003_0' '-Dhadoop.tasklog.totalLogFileSize=0' 'org.apache.hadoop.mapred.Child''211.69.205.3' '50070' 'attempt_201112121911_0001_m_000003_0' '1443792525'  < /dev/null  1>> /home/xhh/workspace/Java/HMapred/build/test/mapred/task/log/userlogs/attempt_201112121911_0001_m_000003_0/stdout 2>> /home/xhh/workspace/Java/HMapred/build/test/mapred/task/log/userlogs/attempt_201112121911_0001_m_000003_0/stderr"

从这个调用系统的语句中我们可以归纳出这样的一个运行语句模式：java  \*\*\*org.apache.hadoop.mapred.Child\*\*\*，其中，\*\*\*表示运行时所需要的一些JVM参数以及应用程序所依赖的第三方jar包和相应的配置参数；\*\*\*表是应用程序运行时本身需要的一些输入参数。因此，TaskTracker节点上用于执行Task的JVM进程的拥有者是作业，也就是说，一个作业只对应一个JVM进程，它的任务会复用这个JVM进程，而其它的作业的Task则只能使用自己所属作业的JVM进程(TaskTracker会为该Job重新开启一个JVM进程)，这样做的目的是为了防止不同的作业之间存在的jar包冲突问题。另外，TaskTracker节点不自己开启线程而是开启进程来执行作业的Task，一方面是为了jar冲突(可以通过ClassLoad来解决)，而主要的是因为作业任务的map/reduce操作被全权交给了用户来执行，这样一来的话，就很有可能发生无法预知的错误，从而影响到TaskTracker节点(在TaskTracker进程下开线程执行作业的Task)，使得TaskTracker节点也跟着宕掉。这样一来，不同的Job占用不同JVM进程，进而对TaskTracker节点上正在执行的Job进程了隔离，而互不影响，但是，Hadoop这样设计也大大地降低了它的整体性能。

[0_1323677956U2c2.gif]: /images/20220617/4a7142ef1e7147cd9912e51dac23ac27.png
[0_1323681988zNe3.gif]: /images/20220617/960d737546ae401fa6d30ea6c5f171a1.png
[0_1323690022Rkm9.gif]: /images/20220617/23ddafb2991f45e7a39cdcf8bb1a2f2d.png
[Java]: http://blog.csdn.net/xhh198781/article/details/6824289