Hadoop MapReduce执行全流程关键代码.doc

HadoopMap/Reduce执行流程关键代码

JobClient.runJob(conf)|运行job
|-->JobClientjc=newJobClient(job);
|-->RunningJobrj=jc.submitJob(job);
|-->submitJobInternal(job);
|-->intreduces=job.getNumReduceTasks();
|-->JobContextcontext=newJobContext(job,jobId);
|-->maps=writeOldSplits(job,submitSplitFile);
|-->job.setNumMapTasks(maps);
|-->job.writeXml(out);
|-->JobStatusstatus=jobSubmitClient.submitJob(jobId);
----------------------------------------------------------------------------------------------

JobTracker.submitJob(JobId)|提交job
|-->JobInProgressjob=newJobInProgress(jobId,this,this.conf);
|-->checkAccess(job,QueueManager.QueueOperation.SUBMIT_JOB);|检查权限
|-->checkMemoryRequirements(job);|检查内存需求
|-->addJob(jobId,job);|添加至job队列
|-->jobs.put(job.getProfile().getJobID(),job);
|-->for(JobInProgressListenerlistener:jobInProgressListeners)｜添加至监听器，供调度使用
|-->listener.jobAdded(job);

------------------------------------------------------------------------------------------------------
JobTracker.heartbeat()|JobTracker启动后供TaskTracker以RPC方式来调用，返回Response集合
|-->List<TaskTrackerAction>actions=newArrayList<TaskTrackerAction>();
|-->tasks=taskScheduler.assignTasks(taskTrackerStatus);|通过调度器选择合适的tasks
|-->for(Tasktask:tasks)
|-->expireLaunchingTasks.addNewTask(task.getTaskID());
|-->actions.add(newLaunchTaskAction(task));|实际actions还会添加commmitTask等
|-->response.setHeartbeatInterval(nextInterval);
|-->response.setActions(actions.toArray(newTaskTrackerAction[actions.size()]));
|-->returnresponse;

---------------------------------------------------------------------------------------------------------

TaskTracker.offerService|TaskTracker启动后通过offerservice()不断发心跳至JobTracker中
|-->transmitHeartBeat()
|-->HeartbeatResponseheartbeatResponse=jobClient.heartbeat(status,justStarted,justInited,askForNewTask,heartbeatResponseId);
|-->TaskTrackerAction[]actions=heartbeatResponse.getActions();
|-->for(TaskTrackerActionaction:actions)
|-->if(actioninstanceofLaunchTaskAction)
|-->addToTaskQueue((LaunchTaskActi