hadoop-map-reduce执行流程调研报告
参与者:
Client 简称:C
JobTracker 简称:JT
TaskTracker 简称:TT
DataNode 简称:DN
NameNode 简称:NN
流程描述:
简述:
作业的提交 -> Map任务的分配和执行 -> Reduce任务的分配和执行 -> 作业的完成
每个具体的任务又分: 准备输入-> 算法的执行 -> 输出结果,三个步骤.
作业的提交:
1. /home/hadoop/hadoop-1.0.3/bin/hadoo jar ~/learning-hadoop.jar com.easou.mr.bookMark.BookMarkMR
2. RunJar.java -> main
3.反射 -BookMarkMR->main
4. org.apache.hadoop.mapreduce.Job ->waitForCompletion
//提交
i f (state == JobState.DEFINE) {
submit();
}
5. org.apache.hadoop.mapred.JobClient -> submitJobInternal
6.
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.jar(执行jar)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.split(输入分片文件)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.splitmetainfo(分片meta)
/data/tmp/core/mapred/staging/hadoop/.staging/jobid/job.xml(jobconf配置信息)
7.
this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
JobID jobId = jobSubmitClient.getNewJobId();//jobid的生成方式
public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts) throws IOException
synchronized (jobs) {
synchronized (taskScheduler) {
jobs.put(job.getProfile().getJobID(), job);
for (JobInProgressListener listener : jobInProgressListeners) {
listener.jobAdded(job);
}
}
}
Map reduce任务的分配和执行:
C rpc 调用 addJob 触发监听:
JobQueueJobInProgressListener(用于监控job的运行状态)
EagerTaskInitializationListener;(用于对Job进行初始化)
public void run() {
JobInProgress job = null;
while (true) {
try {
synchronized (jobInitQueue) {
while (jobInitQueue.isEmpty()) {
jobInitQueue.wait();
}
job = jobInitQueue.remove(0);
}
threadPool.execute(new InitJob(job));
} catch (InterruptedException t) {
LOG.info("JobInitManagerThread interrupted.");
break;
}
}
LOG.info("Shutting down thread pool");
threadPool.shutdownNow();
}
//后台守护运行
jobInitManagerThread.setDaemon(true);
this.jobInitManagerThread.start();
死循环等待job加入,并初始化.
JT初始化task
JobInProgress -> initTasks
1.从HDFS中读取job.split文件从而生成input splits
2.map task的个数就是input split的个数
numMapTasks = splits.length;
3.为每个map tasks生成一个TaskInProgress来处理一个input split
4.对于map task,将其放入nonRunningMapCache,是一个Map<Node, List<TaskInProgress>>,也即对于map task来讲,其将会被分配到其input split所在的Node上。nonRunningMapCache将在JobTracker向TaskTracker分配map task的时候使用。
5.创建reduce task
6.reduce task放入nonRunningReduces,其将在JobTracker向TaskTracker分配reduce task的时候使用。
6.创建两个cleanup task,一个用来清理map,一个用来清理reduce.
7.创建两个初始化 task,一个初始化map,一个初始化reduce.
TT:
1. 一直运行
2. 每隔一段时间就向JobTracker发送heartbeat
3. 在heartbeat中要返回给JobTracker一些统计信息, 报告给JobTracker,此TaskTracker的当前状态
4. 发送Heartbeat到JobTracker,得到response
5. 从Response中得到此TaskTracker需要做的事情
6. 如果是运行一个新的Task,则将Action添加到任务队列中
7. 当满足下面的条件的时候,此TaskTracker请求JobTracker为其分配一个新的Task来运行:
当前TaskTracker正在运行的map task的个数小于可以运行的map task的最大个数
当前TaskTracker正在运行的reduce task的个数小于可以运行的reduce task的最大个数
8. 向JobTracker发送RPC调用Heartbeat
JT 分配 task:
根据数据所处的位置与Task Tracker的距离,有如下几种data locality级别:
0 node-local 输入分片就在Task Tracker本地
1 rack-local 输入分片在Task Tracker所在的rack内其它Task Tracker上
2 off-switch 输入分片在其它的rack内
1.先调度优先级高的作业,统一优先级的作业则先进先出;
2.尽量使集群每一个TaskTracker达到负载均衡(这个均衡是task数量上的而不是实际的工作强度);
3.尽量分配作业的本地任务给TaskTracker,但不是尽快分配作业的本地任务给TaskTracker,最多分配一个非本地任务给TaskTracker(一是保证任务的并发性,二是避免有些TaskTracker的本地任务被偷走),最多分配一个reduce任务;
4..为紧急的Task预留一定的slot;
作业的完成:
Map Clean Up Task(Map任务服务器的清理任务,用于清理相关的过期的文件和环境...) -> Map Setup Task(Map任务服务器的安装任务,负责配置好相关的环境...) -> Map Tasks -> Reduce Clean Up Task -> Reduce Setup Task -> Reduce Tasks
流程图:
整体分配流程:
Map的具体流程
Reduce的具体流程
hadoop-map-reduce执行流程调研报告
猜你喜欢
转载自shixm.iteye.com/blog/1660058
今日推荐
周排行