MapReduce中的JobInProgressListener结构剖析

本帖最后由 hadoopor 于 2009-11-28 08:55 编辑
查看JobTracker类的源代码，可以看到一个JobTracker会关联一到多个JobInProgressListener，如下所示：
public class
[color=]JobTracker

{
    private final List   jobInProgressListeners =
                                       new
[color=]CopyOnWriteArrayList
();
}
那么一个JobTracker会究竟会关联多少个JobInProgressListener了吗？为啥不是一个了吗？这个和具体的TaskScheduler有关。当运用
[color=]JobQueueTaskScheduler
调度器时，结构如下图所示：
也就是一个JobTracker会关联两个JobInProgressListener对象，其中
JobQueueJobInProgressListener
是由JobQueueTaskScheduler调试器实现的，而
[color=]EagerTaskInitializationListener
是由JobTracker提供的，因此EagerTaskInitializationListener
具有通用性，不是一个调度器特有的。

那么这两个JobInProgressListener（JobQueueJobInProgressListener和EagerTaskInitializationListener
）有啥不同了吗？

1. EagerTaskInitializationListener
用来初始化job，它内部有一个线程，专门用来初始化job，将job分解成tasks。提交job时，会将job放入EagerTaskInitializationListener提供的队列jobInitQueue中，而EagerTaskInitializationListener的内部线程JobInitManager则一直监控队列，当从队列中取出一个job时，就创建一个InitJob线程来初始化这个job，每个job都用一个JobInProgress对象来表示。通过job生成tasks的工作是由JobInProgress完成的，可以参见JobInProgress.initTasks实现。

2. JobQueueJobInProgressListener
EagerTaskInitializationListener中的队列是用于job的初始化，在JobQueueJobInProgressListener中也有一个队列jobQueue，这个队列是供调度器运用的，所有需要由调度器调度的job都存在这个队列中。但实际上对一个job的调度包括两个层面：一个对多个job的高度，另一个是对同一个job内各个tasks的调度，在JobQueueJobInProgressListener
只维护了job队列，而在调度tasks时需要利用JobInProgress来完成。

在CapacityTaskScheduler和PriorityScheduler两个调度器中，并没有EagerTaskInitializationListener，但在FairScheduler和JobQueueTaskScheduler两个调度器中均有EagerTaskInitializationListener
。

不知道上面讲得是否够清楚，有空一块来问题吧。

"

是指TaskTracker调用heartbeat()进行任务分派的时候吗吗？

我也想附和這個話題, 有個小問題 --
第一, 並沒有看到JobTracker.java 裡面指定运用EagerTaskInitializationListener 和JobQueueJobInProgressListener 作為它的listeners?
並且, 在宣告
private final List   jobInProgressListeners =
                                       new CopyOnWriteArrayList();
之後, JobTracker 並沒有运用addJobInProgressListener() 添加listeners; 上述的兩種listeners 是從哪裡來的呢?
謝謝! :)

"

我来回答这个
listeners并不是在JobTracker中添加的，它是在TaskScheduler开启的时候被添加进去的，(TaskScheduler手上握有一个JobTracker的引用)代码如下:
org.apache.hadoop.mapred.JobQueueTaskScheduler[ol]

        public synchronized void start() throws IOException {

                super.start();

                taskTrackerManager.addJobInProgressListener(jobQueueJobInProgressListener);

                eagerTaskInitializationListener.setTaskTrackerManager(taskTrackerManager);

                eagerTaskInitializationListener.start();

                taskTrackerManager.addJobInProgressListener(eagerTaskInitializationListener);

        }[/ol]复制代码