FairScheduler job初始化过程源码浅析

485次阅读

共计 16259 个字符，预计需要花费 41 分钟才能阅读完成。

上一篇文章（http://www.linuxidc.com/Linux/2013-12/93701.htm）说到了 jobTracker 中的 submitJob()方法，这个方法最终会调用 listener.jobAdded(job)，将 Job 注册到 TaskScheduler 中，由其进行调度。今天接着研究。Hadoop 中默认的 TaskScheduler 是 JobQueueTaskScheduler，采用的是 FIFO(先进先出)原则进行调度，还有 FiarScheduler 和 CapacityTaskScheduler 两种调度类（非 hadoop 自带，不过 hadoop 也把他们加入到类库中），这两个类可以在 hadoop 目录下的 lib 包下找到，源码在 src/contrib 下可以找到。主要对 FairScheduler 进行解读。

上文提到 jobTracker 最终将 job 注册到 jobListener 中，下面就来看看 FairScheduler 的 JobListener。

1.FairScheduler.JobListener.addJob()：这个方法比较简单，JobSchedulable mapSched = ReflectionUtils.newInstance(conf.getClass(“mapred.jobtracker.jobSchedulable”, JobSchedulable.class, JobSchedulable.class), conf)这里通过反射获得两个 JobSchedulable 对象，也就是默认的 FairScheduler.JobSchedulable 对象，一个是 mapSched，一个是 redSched，然后进行 JobSchedulable 的初始化，比较简单。infos.put(job, info)将 job 添加到 infos（存放所有的 jobInPorgress 对象）中，同时将 job 添加到 PoolScheduable 中，主要是根据配置的 poolName 获取对应的 pool。下面的是重点，update()方法，下面看看这个方法。

public void jobAdded(JobInProgress job) {
synchronized (FairScheduler.this) {
eventLog.log(“JOB_ADDED”, job.getJobID());
JobSchedulable mapSched = ReflectionUtils.newInstance(
conf.getClass(“mapred.jobtracker.jobSchedulable”, JobSchedulable.class,
JobSchedulable.class), conf);
mapSched.init(FairScheduler.this, job, TaskType.MAP);

JobSchedulable redSched = ReflectionUtils.newInstance(
conf.getClass(“mapred.jobtracker.jobSchedulable”, JobSchedulable.class,
JobSchedulable.class), conf);
redSched.init(FairScheduler.this, job, TaskType.REDUCE);

JobInfo info = new JobInfo(mapSched, redSched);
infos.put(job, info);
poolMgr.addJob(job); // Also adds job into the right PoolScheduable
update();
}
}

2.FairScheduler.update()：跳过看不懂的，直接看 poolMgr.reloadAllocsIfNecessary()，这个方法主要是读取 FairScheduler 的配置文件（fair-scheduler.xml），由 mapred.fairscheduler.allocation.file 参数设置，这里是根据配置文件的最后修改时间 +ALLOC_RELOAD_INTERVAL 决定是否重新加载配置文件，加载文件的时候就是简单地读取 xml 文件。接着看 update 方法，加载完配置文件之后会遍历 infos（保存了 FairScheduler 所有的 jobInProgress），遍历的时候去除成功了的 job 和失败了的 job 以及被 kill 掉的 job，同时也会从 pool 中去掉该 job。接下来就是 updateRunnability()，这个方法会根据 userMaxJob 以及 poolMaxJob 数量进行判断是否启动 job。

List<JobInProgress> toRemove = new ArrayList<JobInProgress>();
for (JobInProgress job: infos.keySet()) {
int runState = job.getStatus().getRunState();
if (runState == JobStatus.SUCCEEDED || runState == JobStatus.FAILED
|| runState == JobStatus.KILLED) {
toRemove.add(job);
}
}
for (JobInProgress job: toRemove) {
jobNoLongerRunning(job);
}

3.FairScheduler.updateRunnability()：第一步将所有 infos 中剩余的 job（成功以及失败的任务会在 update 时清除）状态全部设为 notrunning。接着对 infos 中的 job 进行排序，Collections.sort(jobs, new FifoJobComparator())，排序规则是 FIFO 原则（奇怪，不懂）。然后接着对 jobs 进行遍历，同时根据该 job 的提交用户和提交的 pool 的最大提交 job 数量决定是否将其添加到任务队列中（就是两个 list），如果该 job 状态 =RUNNING，则 jobinfo.running=true，如果 job 状态 =PREP（准备中），则对其进行初始化（注意这里只对 job 状态 =RUNNING 和 PREP 的 job 进行操作）。jobInitializer.initJob(jobInfo, job)进行 job 初始化，这里使用到 jdk 的 threadPool（其实就是将 thread 加入到线程池中，由线程池绝对什么时候对其进行执行，总之都会调用 thread 的 run 方法），看看 thread 的 run 方法。run 方法中调用 ttm.initJob(job)，此处的 ttm 就是 jobTracker，现在回到 jobTracker 去。

if (userCount < poolMgr.getUserMaxJobs(user) &&
poolCount < poolMgr.getPoolMaxJobs(pool)) {
if (job.getStatus().getRunState() == JobStatus.RUNNING ||
job.getStatus().getRunState() == JobStatus.PREP) {
userJobs.put(user, userCount + 1);
poolJobs.put(pool, poolCount + 1);
JobInfo jobInfo = infos.get(job);
if (job.getStatus().getRunState() == JobStatus.RUNNING) {
jobInfo.runnable = true;
} else {
// The job is in the PREP state. Give it to the job initializer
// for initialization if we have not already done it.
if (jobInfo.needsInitializing) {
jobInfo.needsInitializing = false;
jobInitializer.initJob(jobInfo, job);
}
}
}
}

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2013-12/93702p2.htm

相关阅读：

Ubuntu 13.04 上搭建 Hadoop 环境 http://www.linuxidc.com/Linux/2013-06/86106.htm

Ubuntu 12.10 +Hadoop 1.2.1 版本集群配置 http://www.linuxidc.com/Linux/2013-09/90600.htm

Ubuntu 上搭建 Hadoop 环境（单机模式 + 伪分布模式）http://www.linuxidc.com/Linux/2013-01/77681.htm

Ubuntu 下 Hadoop 环境的配置 http://www.linuxidc.com/Linux/2012-11/74539.htm

单机版搭建 Hadoop 环境图文教程详解 http://www.linuxidc.com/Linux/2012-02/53927.htm

搭建 Hadoop 环境（在 Winodws 环境下用虚拟机虚拟两个 Ubuntu 系统进行搭建）http://www.linuxidc.com/Linux/2011-12/48894.htm

4.JobTracker.initJob()：主要调用 job.initTasks()，下面进入到 JobInProgress.initTasks()。

5.JobInProgress.initTasks()：为 job 对象设置优先级 setPriority(this.priority)，接着读取分片信息文件获取分片信息，SplitMetaInfoReader.readSplitMetaInfo()这个方就是 jobInPorgress 用来读取分分片信息的，读取过程与写入过程相对应，具体还是较简单的。读取了分片信息之后，根据分片数量创建相应数量的 mapTask（TaskInProgress 对象），接下来会执行 nonRunningMapCache = createCache(splits, maxLevel)，这个方法是根据每个分片的 location 信息，然后根据 location 的 host 判断每个 host 上所有的 job，并放入 cache 中。接着根据设置的 reduce 数量新建对应的 reduceTask（TaskInProgress 对象），并加入到 nonRunningReduces 队列中，并根据 mapred.reduce.slowstart.completed.maps（百分比，默认是 5%）参数的值计算 completedMapsForReduceSlowstart（多少 map 任务完成的时候启动 reduce 任务）。之后就是分别新建两个 setUp 任务和 cheanUp 任务，分别对应 map 和 reduce task。到此 initTask 完成，initTask 完成 JobTracker 的 initJob 也就差不多完成了，接着 FairScheduler 的 updateRunnability()也就完成了。回到 FairScheduler.update()。

6.FairScheduler.update()：

for (Pool pool: poolMgr.getPools()) {
pool.getMapSchedulable().updateDemand();
pool.getReduceSchedulable().updateDemand();
}

// Compute fair shares based on updated demands
List<PoolSchedulable> mapScheds = getPoolSchedulables(TaskType.MAP);
List<PoolSchedulable> reduceScheds = getPoolSchedulables(TaskType.REDUCE);
SchedulingAlgorithms.computeFairShares(
mapScheds, clusterStatus.getMaxMapTasks());
SchedulingAlgorithms.computeFairShares(
reduceScheds, clusterStatus.getMaxReduceTasks());

// Use the computed shares to assign shares within each pool
for (Pool pool: poolMgr.getPools()) {
pool.getMapSchedulable().redistributeShare();
pool.getReduceSchedulable().redistributeShare();
}

if (preemptionEnabled)
updatePreemptionVariables();
}

看不懂，先到这吧，等下次慢慢研究吧。