Hadoop 1.x 版本中的Task Controller 介绍

ewrfdsfwqe

目前网络上对Hadoop 的 TaskController介绍的资料虽然不算少，但感觉纯谈感念的占多数，可能会使读者理解起来感到比较抽象，因此我在这里整理了一些自己对Task Controller的浅见，以及为本文设计的实例，希望能对读者起到一点帮助。

本文介绍Task Controller的顺序为：作用、配置、原理(由表及里、由浅入深)。另外，本文的代码是基于ApacheHadoop-1.1.1。
作用关于Task Controller的作用，在org.apache.hadoop.mapred.TaskController.java的类说明信息中提到：“Controls initialization, finalization and clean up oftasks, and also the launching and killing of task JVMs”。从此处不难看出Task Controller的作用：控制task的初始化、终止以及执行环境清理，同时控制运行task所需的JVMs的启动和终止。简单说，它控制了一个task具体怎么被执行，这很好理解。TaskController.java只是一个abstract class，而它的具体实现类有两个：DefaultTaskController和LinuxTaskController.java：


DefaultTaskController和LinuxTaskController.java有什么区别？他们最主要的区别主要是从安全性上考虑的——以哪个user的身份操作OS的文件（请注意：不是针对HDFS，而是OS）：
-DefaultTaskController：无论哪个用户提交Job，所有的task都以task tracker daemon的拥有者的身份运行。这种情况下，Job的owner还是Job的提交者，并不会因此变成task tracker daemon的拥有者，而且Job对HDFS的操作也是以Job的提交者的身份进行的。但是，task中所有对OS的操作都是以task tracker daemon的拥有者的身份运行，而task tracker daemon的拥有者往往在OS上拥有较高的权限，这样就带来了较大的安全隐患：提交Job的用户，可以利用tasktracker daemon的拥有者的身份权限来对OS做出较大的破坏行为。
- LinuxTaskController：无论哪个用户提交Job，所有的task都以提交Job的用户的身份运行。与DefaultTaskController相同的是， Job的owner是Job的提交者，并且Job对HDFS的操作也是以Job的提交者的身份进行。但不同的是，task中所有对OS的操作都是以Job的提交者的身份运行，这样就屏蔽了DefaultTaskController所拥有的安全隐患。

接下来，通过一个具体的例子来验证以上的结论。
首先，设计一个简单的MapReduce Job：该Job没有指定的Reducer，只指定一个名为FOMapper的Mapper。FOMapper的作用是将一个HDFS上的文件拷贝到本地OS的/home/tom文件夹下，而/home/tom的拥有者为用户tom。因此，非tom用户和非特权用户并不能在/home/tom目录下创建文件。测试所用Hadoop cluster的tasktracker daemon的拥有者为tom 用户。

本MapReduce Job的代码如下：

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

publicclass TaskControllerTest {

    publicstaticvoid copyFromHDFSToLocal(Configuration conf)

           throws IOException {

       FileSystem hdfs = FileSystem.get(conf);

       Path srcPath = new Path("/user/tom/test.txt");

       Path dstPath = new Path("/home/tom/test.txt");

       hdfs.copyToLocalFile(srcPath, dstPath);

    }

    publicstaticclass FOMapperextends

           Mapper<Object, Text, Text, LongWritable> {

       @Override

       publicvoid map(Object key, Text value, Context context)

              throws IOException, InterruptedException {

           copyFromHDFSToLocal(context.getConfiguration());

           context.write(value, new LongWritable(1));

       }

    }

    /**

    * @param args

    * @throws IOException

    * @throwsClassNotFoundException

    * @throwsInterruptedException

    */

    publicstaticvoid main(String[] args)throws IOException,InterruptedException, ClassNotFoundException {

       Configuration conf = new Configuration();

       Job job = new Job(conf,"FileOperations Job");

       job.setMapperClass(FOMapper.class);

       job.setJarByClass(FOMapper.class);

      

       job.setMapOutputKeyClass(Text.class);

       job.setMapOutputValueClass(LongWritable.class);

       FileInputFormat.setInputPaths(job,new Path(args[0]));

       FileOutputFormat.setOutputPath(job,new Path(args[1]));

       System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

把此MapReduce Job打包，然后分别在以下四种情况下运行，并验证结果（用户可以自己在本地进行类似测试）：
a) Hadoop cluster配置使用DefaultTaskController（具体配置方法请参考下文“配置”部分），然后以用户tom来提交Job。
结果：Job的拥有者为tom，并成功运行，且能够将HDFS上的文件成功拷贝到OS的/home/tom文件夹下面，而且该文件在OS的拥有者是tom。

b) Hadoop cluster配置使用DefaultTaskController，然后以用户jerry(非tom用户和非特权用户)来提交Job。
结果：Job的拥有者为jerry，并成功运行，且能够将HDFS上的文件成功拷贝到OS的/home/tom文件夹下面，而且该文件在OS的拥有者并非jerry，而是tom。

c) Hadoop cluster配置使用LinuxTaskController（具体配置方法请参考下文“配置”部分），然后以用户tom来提交Job。
结果：Job的拥有者为tom，并成功运行，且能够将HDFS上的文件成功拷贝到OS的/home/tom文件夹下面，而且该文件在OS的拥有者是tom。

d) Hadoop cluster配置使用LinuxTaskController，然后以用户jerry(非tom用户和非特权用户)来提交Job。
结果：Job的拥有者为jerry，但Job执行失败，并抛出“Permission Denied”的异常——这个结果是能够被预料到的，并且是正确的：在使用LinuxTaskController的时候，任何对OS的文件操作都是以Job的提交者身份来进行。所以当Job的提交者是jerry时，在task试图以jerry用户的身份向/home/tom目录下写文件时就会碰到“PermissionDenied”的错误。这也规避了DefaultTaskController所拥有的安全风险。

从上面的试验不难看出LinuxTaskController比DefaultTaskController拥有更高的安全性，因此也推荐用户在Hadoop cluster中使用LinuxTaskController。

配置A. 配置使用DefaultTaskController：
无需做任何配置，因为Hadoop默认使用的TaskController就是DefaultTaskController。

B. 配置使用LinuxTaskController：
a) task-controller：

这是一个setuid的可执行二进制文件，Hadoop用它来以提交Job的用户身份来执行task。Hadoop自带该文件，它默认的位置为${HADOOP_HOME}/bin/ task-controller，而且所属用户必须是root，所属用户组必须由参数mapreduce.tasktracker.group指定（在mapred-site.xml和task-controller.cfg中指定）。另外，它的读写权限一般是4754 或者-rwsr-xr--。


b) mapred-site.xml：
在mapred-site.xml中添加如下配置：
  <property>
   <name>mapred.task.tracker.task-controller</name>
   <value>org.apache.hadoop.mapred.LinuxTaskController</value>
  </property>
  <property>
   <name>mapreduce.tasktracker.group</name>
    <value>tom</value>
  </property>
注意：
- mapred.task.tracker.task-controller：指定Hadoop选用的Task Contoller类型。默认是org.apache.hadoop.mapred.DefaultTaskController。当使用LinuxTaskController时，需填org.apache.hadoop.mapred.LinuxTaskController。
- mapreduce.tasktracker.group：定义task-controller 文件所属的用户组。如果task-controller的用户组不符合该设定，则会在启动hadoop task tracker的时候就会抛出错误：“[ERROR] tasktracker failed to start”。

c) task-controller.cfg：
保存task controller的具体配置。该文件默认的位置是的${HADOOP_HOME}/conf/ task-controller.cfg，但其位置也可由环境变量HADOOP_SECURITY_CONF_DIR显示地定义：“export HADOOP_SECURITY_CONF_DIR=/var/task-controller-conf”。taskcontroller.cfg 文件的属性为400。其内容一般如下：
mapred.tasktracker.tasks.sleeptime-before-sigkill=#sleeptime before sig kill is to be sent to process group after sigterm is sent.Should be in seconds
hadoop.log.dir=/var/hadoop/logs
mapred.local.dir=/hadoop/mapred/local
mapreduce.tasktracker.group=hadoop
min.user.id=100
banned.users=foo,bar
说明：
- min.user.id：主要是用来限制一些高权限的用户执行task
- banned.users：用来限制一些特定的用户执行task
另外，Hadoop要求HADOOP_SECURITY_CONF_DIR路径所包含的各级目录仅能对root用户可写。一般情况下，HADOOP_SECURITY_CONF_DIR路径所包含的各级目录的所属用户为root、所属用户组保持和TaskTracker的用户组一致，而目录的属性可设置为755。


原理知道了以上Task Controller的功能后，我们再来探究它的实现原理就容易很多。不过本文不打算事无巨细地罗列代码及代码间的调用关系(因为这些可以在代码中很容易地找到)，只谈几个笔者认为值得一提的要点：
A. Hadoop框架中的什么地方调用过Task Controller？
在Task执行的各个阶段，与Task执行相关的很多地方都会调用Task Controller的不同方法。比如TaskTracker.java、TaskRunner.java、JvmManager.java与UserLogManager.java等。

B. DefaultTaskController.java与LinuxTaskController.java实现上的主要不同？
抽象类TaskController.java中定义了一些抽象方法，主要的是setup()、initializeJob()、launchTask()、signalTask()、createLogDir()以及deleteAsUser()等。DefaultTaskController.java与LinuxTaskController.java分别对这些方法进行了自己的实现。而实现上主要的不同是：
- 在DefaultTaskController.java中，其方法里面的各种操作主要是直接调用Hadoop FileSystem API或者是Java IO API去完成，也就是由DefaultTaskController.java自己实现具体的行为。比如deleteAsUser()方法：

  publicvoiddeleteAsUser(Stringuser,

                           String subDir)throws IOException {

    String dir = TaskTracker.getUserDir(user)+ Path.SEPARATOR + subDir;

    for(Path fullDir:allocator.getAllLocalPathsToRead(dir, getConf())) {

      fs.delete(fullDir,true);

    }

  }
- 在LinuxTaskController.java中，其具体的操作主要靠调用task-controller这个可执行二进制文件里的方法来实现。以下依然拿deleteAsUser()方法来举例：

  // Path tothesetuid executable.

  private StringtaskControllerExe;

  privatestaticfinal StringTASK_CONTROLLER_EXEC_KEY =

    "mapreduce.tasktracker.task-controller.exe";

  

  @Override

  publicvoid setConf(Configuration conf) {

    super.setConf(conf);

   FilehadoopBin = new File(System.getenv("HADOOP_HOME"),"bin");

    String defaultTaskController =

        new File(hadoopBin,"task-controller").getAbsolutePath();

    taskControllerExe = conf.get(TASK_CONTROLLER_EXEC_KEY,

                                 defaultTaskController);      

  }
... ...

  @Override

  publicvoiddeleteLogAsUser(Stringuser, String subDir)throws IOException {

    String[] command =

      new String[]{taskControllerExe,

                   user,

                   localStorage.getDirsString(),

                   Integer.toString(Commands.DELETE_LOG_AS_USER.getValue()),

                   subDir};

    ShellCommandExecutor shExec =newShellCommandExecutor(command);

    if (LOG.isDebugEnabled()) {

      LOG.debug("deleteLogAsUser: " + Arrays.toString(command));

    }

    shExec.execute();

  }

C. 可执行二进制文件task-controller是怎么生成的？
task-controller由c++代码生成，具体的代码位于Hadoop工程下src/c++/task-controller文件夹下，如下图所示：



task-controller的核心代码在蓝色方框内，有兴趣的读者可以自己看看这些代码。这里主要对两个最重要的文件做个简介：
- main.c：task-controller的入口程序，接受、检验命令行命令，并调用task-controller.c里面所实现的方法。
- task-controller.c：task-controller所需方法的具体实现。它里面定义了大量的方法：


值得一提的是，用户可以根据需要，通过自己改写、丰富task-controller.c、task-controller.h以及main.c等代码文件来拓展task-controller的功能。笔者就曾经通过改写task-controller来解决了Hadoop的一个问题，并把代码贡献回给了Apache Hadoop社区（https://issues.apache.org/jira/browse/MAPREDUCE-4490）。