现在的位置: 首页 > 综合 > 正文

hadoop中槽-slot是线程还是进程讨论

2017年11月19日 ⁄ 综合 ⁄ 共 3914字 ⁄ 字号 评论关闭

hadoop一个节点默认起两个map slot,这两个slot是多线程吗?

hadoop-0.21.0 源码中是这样的:

首先看看 org.apache.hadoop.mapred.TaskTracker 类:
=====================================================================================
> 两个类变量 maxMapSlots 和 maxReduceSlots:

  1. maxMapSlots = conf.getInt(TT_MAP_SLOTS, 2);
  2.     maxReduceSlots = conf.getInt(TT_REDUCE_SLOTS, 2);

复制代码

  其中

  1.    public static final String TT_MAP_SLOTS =  "mapreduce.tasktracker.map.tasks.maximum";
  2.     public static final String TT_REDUCE_SLOTS = "mapreduce.tasktracker.reduce.tasks.maximum";

复制代码


> 类方法 initializeMemoryManagement() 中 ,根据 slots 来决定申请内存的大小

  1. totalMemoryAllottedForTasks =
  2.         maxMapSlots * mapSlotMemorySizeOnTT + maxReduceSlots
  3.             * reduceSlotSizeMemoryOnTT;

复制代码


> 类方法 TaskTracker.initialize() 中会起两个 TaskLauncher 线程,分别负责启动 Mapper 和 Reduce 任务:

  1.     mapLauncher = new TaskLauncher(TaskType.MAP, maxMapSlots);
  2.     reduceLauncher = new TaskLauncher(TaskType.REDUCE, maxReduceSlots);

复制代码

=====================================================================================
再看看 org.apache.hadoop.mapred.TaskTracker.TaskLauncher 类,它负责启动 Mapper/Reducer 任务。

> 初始化 TaskLauncher 时,需要传入 slots 的数量:
-------------------------------------------

  1.     public TaskLauncher(TaskType taskType, int numSlots) {
  2.       ... ...
  3.       this.maxSlots = numSlots;
  4.       this.numFreeSlots = new IntWritable(numSlots);
  5.       ... ...
  6.     }

复制代码

  特别要注意 numFreeSlots 这个类变量:
      private IntWritable numFreeSlots;
-------------------------------------------
> TaskLauncher.run() 中,循环地看是否有新的 Task 需要启动,并且看是否有足够的 slots 可用:

  1.   while () {
  2.     while (numFreeSlots.get() < task.getNumSlotsRequired()) {
  3.        .......
  4.     }
  5.     numFreeSlots.set(numFreeSlots.get() - task.getNumSlotsRequired());   // 用完了就减掉
  6.   }

复制代码


> Task 执行完了以后,需要释放 slots :

  1.     public void addFreeSlots(int numSlots) {
  2.       ... ...
  3.       numFreeSlots.set(numFreeSlots.get() + numSlots);
  4.       ... ...
  5.     }

复制代码

=====================================================================================

综合上面看, slots 只是一个逻辑值 ( org.apache.hadoop.mapred.TaskTracker.TaskLauncher.numFreeSlots ),而不是对应着一个线程或者进程。TaskLauncher 会维护这个值,以保证资源使用在控制范围内。

代码可见 : org.apache.hadoop.mapred.TaskTracker.TaskLauncher.run()  。

Mapper 和 Reducer 都是单独的进程,但是它们与 slots 的关系是这样的:

  1. org.apache.hadoop.mapred.TaskTracker.TaskLauncher.run() {
  2.           ... ...
  3.           //got a free slot. launch the task
  4.           startNewTask(tip);
  5.           ... ...
  6. }

复制代码


这里的 slots 有点类似 “令牌” 的感觉:申请资源,先获得令牌;释放资源,交还令牌。

> mapper 和 reducer 都是单独的进程?好像有点不对,是单独的线程吧?

是单独的进程。

启动Mapper/Reducer的总的调用路径是:

  1. org.apache.hadoop.mapred.TaskTracker.TaskLauncher.run()
  2. ->
  3. org.apache.hadoop.mapred.TaskTracker.startNewTask()
  4. ->
  5. org.apache.hadoop.mapred.TaskTracker.launchTaskForJob()
  6. ->
  7. org.apache.hadoop.mapred.TaskTracker.TaskInProgress.launchTask()
  8. ->
  9. org.apache.hadoop.mapred.Task.createRunner()  //  抽象方法,具体实现在子类 MapTask 和 ReduceTask 中
  10.    |-> org.apache.hadoop.mapred.MapTask.createRunner()  // 创建 MapTaskRunner 类实例
  11.    |-> org.apache.hadoop.mapred.ReduceTask.createRunner()  // 创建 ReduceTaskRunner 类实例

复制代码

最终,跟踪到了
MapTaskRunner 和 ReduceTaskRunner 这两个类。

至此,我们看看它们的父类 org.apache.hadoop.mapred.TaskRunner ,以下是类的说明:

  1. /** Base class that runs a task in a separate process.  Tasks are run in a
  2. * separate process in order to isolate the map/reduce system code from bugs in
  3. * user supplied map and reduce functions.
  4. */

复制代码

TaskRunner 虽然 extends Thread (看起来是个线程),但是真正启动Mapper和Reduce进程的代码在函数 TaskRunner.run() 中:

  1.   public final void run() {
  2.       ... ...
  3.       launchJvmAndWait(setup, vargs, stdout, stderr, logSize, workDir, env);
  4.       ... ...
  5.   }

复制代码

其调用了 TaskRunner.launchJvmAndWait() 方法(在此之前还有些创建文件夹、设置配置参数和环境变量等准备性的操作):

  1.   void launchJvmAndWait(List<String> setup, Vector<String> vargs, File stdout,
  2.       File stderr, long logSize, File workDir, Map<String, String> env)
  3.       throws InterruptedException {
  4.     jvmManager.launchJvm(this, jvmManager.constructJvmEnv(setup, vargs, stdout,
  5.         stderr, logSize, workDir, env, conf));
  6.     synchronized (lock) {
  7.       while (!done) {
  8.         lock.wait();
  9.       }
  10.     }
  11.   }

复制代码

上面代码主要是 launch 一个
java
虚拟机进程。这也是Hadoop启动代价很高的原因,因为launch虚拟机是比较耗资源的;于是又提供了Task
JVM Reuse机制。

单独起进程的原因也说得很清楚,就是: isolate the map/reduce system code from bugs in user supplied map and reduce functions。其实就是,通过使用不同的进程空间,进行隔离,防止用户提供的代码中有bug死掉后,造成 TaskTracker  所在进程也死掉(这个死掉了,效果就跟阿凡达里面的发光树被毁了一样)。

Hadoop-0.20.2源码中的实现基本也是差不多的。

抱歉!评论已关闭.