hadoop-map-reduce执行流程调研报告.docx

hadoop-map-reduce执行流程调研报告参与者:Client 简称:CJobTracker 简称:JTTaskTracker 简称:TTDataNode 简称:DNNameNode 简称: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 IOExceptionsynchronized (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 splits2.map task的个数就是input split的个数 numMapTasks = splits.length;3.为每个map tasks生成一个TaskInProgress来处理一个input split4.对于map