Hadoop 学习总结之四：Map-Reduce的过程解析（转载）

深碍√TFBOYSˉ_ 2022-03-29 07:12 93阅读 0赞

# 一、客户端 #

Map-Reduce的过程首先是由客户端提交一个任务开始的。

提交任务主要是通过JobClient.runJob(JobConf)静态函数实现的：

<table> 
 <tbody> 
 <tr> 
 <td> public static RunningJob runJob(JobConf job) throws IOException { &nbsp; //首先生成一个JobClient对象 &nbsp; JobClient jc = new JobClient(job); &nbsp; …… &nbsp; //调用submitJob来提交一个任务 &nbsp; running = jc.submitJob(job); &nbsp; JobID jobId = running.getID(); &nbsp; …… &nbsp; while (true) { &nbsp;&nbsp;&nbsp;&nbsp; //while循环中不断得到此任务的状态，并打印到客户端console中 &nbsp; } &nbsp; return running; } </td> 
 </tr> 
 </tbody> 
</table>

其中JobClient的submitJob函数实现如下：

<table> 
 <tbody> 
 <tr> 
 <td> public RunningJob submitJob(JobConf job) throws FileNotFoundException, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; InvalidJobConfException, IOException { &nbsp; //从JobTracker得到当前任务的id &nbsp; JobID jobId = jobSubmitClient.getNewJobId(); &nbsp; //准备将任务运行所需要的要素写入HDFS： &nbsp; //任务运行程序所在的jar封装成job.jar &nbsp; //任务所要处理的input split信息写入job.split &nbsp; //任务运行的配置项汇总写入job.xml &nbsp; Path submitJobDir = new Path(getSystemDir(), jobId.toString()); &nbsp; Path submitJarFile = new Path(submitJobDir, "job.jar"); &nbsp; Path submitSplitFile = new Path(submitJobDir, "job.split"); &nbsp; //此处将-libjars命令行指定的jar上传至HDFS &nbsp; configureCommandLineOptions(job, submitJobDir, submitJarFile); &nbsp; Path submitJobFile = new Path(submitJobDir, "job.xml"); &nbsp; …… &nbsp; //通过input format的格式获得相应的input split，默认类型为FileSplit &nbsp; InputSplit[] splits = &nbsp;&nbsp;&nbsp; job.getInputFormat().getSplits(job, job.getNumMapTasks()); &nbsp; &nbsp; // 生成一个写入流，将input split得信息写入job.split文件 &nbsp; FSDataOutputStream out = FileSystem.create(fs, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; submitSplitFile, new FsPermission(JOB_FILE_PERMISSION)); &nbsp; try { &nbsp;&nbsp;&nbsp; //写入job.split文件的信息包括：split文件头，split文件版本号，split的个数，接着依次写入每一个input split的信息。 &nbsp;&nbsp;&nbsp; //对于每一个input split写入：split类型名(默认FileSplit)，split的大小，split的内容(对于FileSplit，写入文件名，此split在文件中的起始位置)，split的location信息(即在那个DataNode上)。 &nbsp;&nbsp;&nbsp; writeSplitsFile(splits, out); &nbsp; } finally { &nbsp;&nbsp;&nbsp; out.close(); &nbsp; } &nbsp; job.set("mapred.job.split.file", submitSplitFile.toString()); &nbsp; //根据split的个数设定map task的个数 &nbsp; job.setNumMapTasks(splits.length); &nbsp; // 写入job的配置信息入job.xml文件&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp; out = FileSystem.create(fs, submitJobFile, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new FsPermission(JOB_FILE_PERMISSION)); &nbsp; try { &nbsp;&nbsp;&nbsp; job.writeXml(out); &nbsp; } finally { &nbsp;&nbsp;&nbsp; out.close(); &nbsp; } &nbsp; //真正的调用JobTracker来提交任务 &nbsp; JobStatus status = jobSubmitClient.submitJob(jobId); &nbsp; …… } </td> 
 </tr> 
 </tbody> 
</table>

# 二、JobTracker #

JobTracker作为一个单独的JVM运行，其运行的main函数主要调用有下面两部分：

*  调用静态函数startTracker(new JobConf())创建一个JobTracker对象
 *  调用JobTracker.offerService()函数提供服务

在JobTracker的构造函数中，会生成一个taskScheduler成员变量，来进行Job的调度，默认为JobQueueTaskScheduler，也即按照FIFO的方式调度任务。

在offerService函数中，则调用taskScheduler.start()，在这个函数中，为JobTracker(也即taskScheduler的taskTrackerManager)注册了两个Listener：

*  JobQueueJobInProgressListener jobQueueJobInProgressListener用于监控job的运行状态
 *  EagerTaskInitializationListener eagerTaskInitializationListener用于对Job进行初始化

EagerTaskInitializationListener中有一个线程JobInitThread，不断得到jobInitQueue中的JobInProgress对象，调用JobInProgress对象的initTasks函数对任务进行初始化操作。

在上一节中，客户端调用了JobTracker.submitJob函数，此函数首先生成一个JobInProgress对象，然后调用addJob函数，其中有如下的逻辑：

<table> 
 <tbody> 
 <tr> 
 <td> synchronized (jobs) { &nbsp; synchronized (taskScheduler) { &nbsp;&nbsp;&nbsp; jobs.put(job.getProfile().getJobID(), job); &nbsp;&nbsp;&nbsp; //对JobTracker的每一个listener都调用jobAdded函数 &nbsp;&nbsp;&nbsp; for (JobInProgressListener listener : jobInProgressListeners) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; listener.jobAdded(job); &nbsp;&nbsp;&nbsp; } &nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

EagerTaskInitializationListener的jobAdded函数就是向jobInitQueue中添加一个JobInProgress对象，于是自然触发了此Job的初始化操作，由JobInProgress得initTasks函数完成：

<table> 
 <tbody> 
 <tr> 
 <td> public synchronized void initTasks() throws IOException { &nbsp; …… &nbsp; //从HDFS中读取job.split文件从而生成input splits &nbsp; String jobFile = profile.getJobFile(); &nbsp; Path sysDir = new Path(this.jobtracker.getSystemDir()); &nbsp; FileSystem fs = sysDir.getFileSystem(conf); &nbsp; DataInputStream splitFile = &nbsp;&nbsp;&nbsp; fs.open(new Path(conf.get("mapred.job.split.file"))); &nbsp; JobClient.RawSplit[] splits; &nbsp; try { &nbsp;&nbsp;&nbsp; splits = JobClient.readSplitFile(splitFile); &nbsp; } finally { &nbsp;&nbsp;&nbsp; splitFile.close(); &nbsp; } &nbsp; //map task的个数就是input split的个数 &nbsp; numMapTasks = splits.length; &nbsp; //为每个map tasks生成一个TaskInProgress来处理一个input split &nbsp; maps = new TaskInProgress[numMapTasks]; &nbsp; for(int i=0; i &lt; numMapTasks; ++i) { &nbsp;&nbsp;&nbsp; inputLength += splits[i].getDataLength(); &nbsp;&nbsp;&nbsp; maps[i] = new TaskInProgress(jobId, jobFile, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; splits[i], &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobtracker, conf, this, i); &nbsp; } &nbsp; //对于map task，将其放入nonRunningMapCache，是一个Map&lt;Node, List&lt;TaskInProgress&gt;&gt;，也即对于map task来讲，其将会被分配到其input split所在的Node上。nonRunningMapCache将在JobTracker向TaskTracker分配map task的时候使用。 &nbsp; if (numMapTasks &gt; 0) { &nbsp;&nbsp;&nbsp; nonRunningMapCache = createCache(splits, maxLevel); &nbsp; } &nbsp; //创建reduce task &nbsp; this.reduces = new TaskInProgress[numReduceTasks]; &nbsp; for (int i = 0; i &lt; numReduceTasks; i++) { &nbsp;&nbsp;&nbsp; reduces[i] = new TaskInProgress(jobId, jobFile, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numMapTasks, i, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobtracker, conf, this); &nbsp;&nbsp;&nbsp; //reduce task放入nonRunningReduces，其将在JobTracker向TaskTracker分配reduce task的时候使用。 &nbsp;&nbsp;&nbsp; nonRunningReduces.add(reduces[i]); &nbsp; } &nbsp; //创建两个cleanup task，一个用来清理map，一个用来清理reduce. &nbsp; cleanup = new TaskInProgress[2]; &nbsp; cleanup[0] = new TaskInProgress(jobId, jobFile, splits[0], &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobtracker, conf, this, numMapTasks); &nbsp; cleanup[0].setJobCleanupTask(); &nbsp; cleanup[1] = new TaskInProgress(jobId, jobFile, numMapTasks, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numReduceTasks, jobtracker, conf, this); &nbsp; cleanup[1].setJobCleanupTask(); &nbsp; //创建两个初始化 task，一个初始化map，一个初始化reduce. &nbsp; setup = new TaskInProgress[2]; &nbsp; setup[0] = new TaskInProgress(jobId, jobFile, splits[0], &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobtracker, conf, this, numMapTasks + 1 ); &nbsp; setup[0].setJobSetupTask(); &nbsp; setup[1] = new TaskInProgress(jobId, jobFile, numMapTasks, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numReduceTasks + 1, jobtracker, conf, this); &nbsp; setup[1].setJobSetupTask(); &nbsp; tasksInited.set(true);//初始化完毕 &nbsp; …… } </td> 
 </tr> 
 </tbody> 
</table>

# 三、TaskTracker #

TaskTracker也是作为一个单独的JVM来运行的，在其main函数中，主要是调用了new TaskTracker(conf).run()，其中run函数主要调用了：

<table> 
 <tbody> 
 <tr> 
 <td> State offerService() throws Exception { &nbsp; long lastHeartbeat = 0; &nbsp; //TaskTracker进行是一直存在的 &nbsp; while (running &amp;&amp; !shuttingDown) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long now = System.currentTimeMillis(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //每隔一段时间就向JobTracker发送heartbeat &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long waitTime = heartbeatInterval - (now - lastHeartbeat); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (waitTime &gt; 0) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; synchronized(finishedCount) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (finishedCount[0] == 0) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; finishedCount.wait(waitTime); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; finishedCount[0] = 0; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //发送Heartbeat到JobTracker，得到response &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; HeartbeatResponse heartbeatResponse = transmitHeartBeat(now); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp; //从Response中得到此TaskTracker需要做的事情 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; TaskTrackerAction[] actions = heartbeatResponse.getActions(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (actions != null){ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for(TaskTrackerAction action: actions) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (action instanceof LaunchTaskAction) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //如果是运行一个新的Task，则将Action添加到任务队列中 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; addToTaskQueue((LaunchTaskAction)action); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } else if (action instanceof CommitTaskAction) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; CommitTaskAction commitAction = (CommitTaskAction)action; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (!commitResponses.contains(commitAction.getTaskID())) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; commitResponses.add(commitAction.getTaskID()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } else { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tasksToCleanup.put(action); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; return State.NORMAL; } </td> 
 </tr> 
 </tbody> 
</table>

其中transmitHeartBeat主要逻辑如下：

<table> 
 <tbody> 
 <tr> 
 <td> private HeartbeatResponse transmitHeartBeat(long now) throws IOException { &nbsp; //每隔一段时间，在heartbeat中要返回给JobTracker一些统计信息 &nbsp; boolean sendCounters; &nbsp; if (now &gt; (previousUpdate + COUNTER_UPDATE_INTERVAL)) { &nbsp;&nbsp;&nbsp; sendCounters = true; &nbsp;&nbsp;&nbsp; previousUpdate = now; &nbsp; } &nbsp; else { &nbsp;&nbsp;&nbsp; sendCounters = false; &nbsp; } &nbsp; …… &nbsp; //报告给JobTracker，此TaskTracker的当前状态 &nbsp; if (status == null) { &nbsp;&nbsp;&nbsp; synchronized (this) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; status = new TaskTrackerStatus(taskTrackerName, localHostname, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; httpPort, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; cloneAndResetRunningTaskStatuses( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; sendCounters), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; failures, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; maxCurrentMapTasks, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; maxCurrentReduceTasks); &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; …… &nbsp; //当满足下面的条件的时候，此TaskTracker请求JobTracker为其分配一个新的Task来运行： &nbsp; //当前TaskTracker正在运行的map task的个数小于可以运行的map task的最大个数 &nbsp; //当前TaskTracker正在运行的reduce task的个数小于可以运行的reduce task的最大个数 &nbsp; boolean askForNewTask; &nbsp; long localMinSpaceStart; &nbsp; synchronized (this) { &nbsp;&nbsp;&nbsp; askForNewTask = (status.countMapTasks() &lt; maxCurrentMapTasks || &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; status.countReduceTasks() &lt; maxCurrentReduceTasks) &amp;&amp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; acceptNewTasks; &nbsp;&nbsp;&nbsp; localMinSpaceStart = minSpaceStart; &nbsp; } &nbsp; …… &nbsp; //向JobTracker发送heartbeat，这是一个RPC调用 &nbsp; HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; justStarted, askForNewTask, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; heartbeatResponseId); &nbsp; …… &nbsp; return heartbeatResponse; } </td> 
 </tr> 
 </tbody> 
</table>

# 四、JobTracker #

当JobTracker被RPC调用来发送heartbeat的时候，JobTracker的heartbeat(TaskTrackerStatus status,boolean initialContact, boolean acceptNewTasks, short responseId)函数被调用：

<table> 
 <tbody> 
 <tr> 
 <td> public synchronized HeartbeatResponse heartbeat(TaskTrackerStatus status, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean initialContact, boolean acceptNewTasks, short responseId) &nbsp; throws IOException { &nbsp; …… &nbsp; String trackerName = status.getTrackerName(); &nbsp; …… &nbsp; short newResponseId = (short)(responseId + 1); &nbsp; …… &nbsp; HeartbeatResponse response = new HeartbeatResponse(newResponseId, null); &nbsp; List&lt;TaskTrackerAction&gt; actions = new ArrayList&lt;TaskTrackerAction&gt;(); &nbsp; //如果TaskTracker向JobTracker请求一个task运行 &nbsp; if (acceptNewTasks) { &nbsp;&nbsp;&nbsp; TaskTrackerStatus taskTrackerStatus = getTaskTracker(trackerName); &nbsp;&nbsp;&nbsp; if (taskTrackerStatus == null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; LOG.warn("Unknown task tracker polling; ignoring: " + trackerName); &nbsp;&nbsp;&nbsp; } else { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //setup和cleanup的task优先级最高 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; List&lt;Task&gt; tasks = getSetupAndCleanupTasks(taskTrackerStatus); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (tasks == null ) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //任务调度器分配任务 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tasks = taskScheduler.assignTasks(taskTrackerStatus); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (tasks != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (Task task : tasks) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //将任务放入actions列表，返回给TaskTracker &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; expireLaunchingTasks.addNewTask(task.getTaskID()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; actions.add(new LaunchTaskAction(task)); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; …… &nbsp; int nextInterval = getNextHeartbeatInterval(); &nbsp; response.setHeartbeatInterval(nextInterval); &nbsp; response.setActions( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; actions.toArray(new TaskTrackerAction[actions.size()])); &nbsp; …… &nbsp; return response; } </td> 
 </tr> 
 </tbody> 
</table>

默认的任务调度器为JobQueueTaskScheduler，其assignTasks如下：

<table> 
 <tbody> 
 <tr> 
 <td> public synchronized List&lt;Task&gt; assignTasks(TaskTrackerStatus taskTracker) &nbsp;&nbsp;&nbsp; throws IOException { &nbsp; ClusterStatus clusterStatus = taskTrackerManager.getClusterStatus(); &nbsp; int numTaskTrackers = clusterStatus.getTaskTrackers(); &nbsp; Collection&lt;JobInProgress&gt; jobQueue =　jobQueueJobInProgressListener.getJobQueue(); &nbsp; int maxCurrentMapTasks = taskTracker.getMaxMapTasks(); &nbsp; int maxCurrentReduceTasks = taskTracker.getMaxReduceTasks(); &nbsp; int numMaps = taskTracker.countMapTasks(); &nbsp; int numReduces = taskTracker.countReduceTasks(); &nbsp; //计算剩余的map和reduce的工作量：remaining &nbsp; int remainingReduceLoad = 0; &nbsp; int remainingMapLoad = 0; &nbsp; synchronized (jobQueue) { &nbsp;&nbsp;&nbsp; for (JobInProgress job : jobQueue) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (job.getStatus().getRunState() == JobStatus.RUNNING) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int totalMapTasks = job.desiredMaps(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int totalReduceTasks = job.desiredReduces(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; remainingMapLoad += (totalMapTasks - job.finishedMaps()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; remainingReduceLoad += (totalReduceTasks - job.finishedReduces()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; //计算平均每个TaskTracker应有的工作量，remaining/numTaskTrackers是剩余的工作量除以TaskTracker的个数。 &nbsp; int maxMapLoad = 0; &nbsp; int maxReduceLoad = 0; &nbsp; if (numTaskTrackers &gt; 0) { &nbsp;&nbsp;&nbsp; maxMapLoad = Math.min(maxCurrentMapTasks, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (int) Math.ceil((double) remainingMapLoad / &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numTaskTrackers)); &nbsp;&nbsp;&nbsp; maxReduceLoad = Math.min(maxCurrentReduceTasks, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (int) Math.ceil((double) remainingReduceLoad &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; / numTaskTrackers)); &nbsp; } &nbsp; …… &nbsp; //map优先于reduce，当TaskTracker上运行的map task数目小于平均的工作量，则向其分配map task &nbsp; if (numMaps &lt; maxMapLoad) { &nbsp;&nbsp;&nbsp; int totalNeededMaps = 0; &nbsp;&nbsp;&nbsp; synchronized (jobQueue) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (JobInProgress job : jobQueue) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (job.getStatus().getRunState() != JobStatus.RUNNING) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; continue; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Task t = job.obtainNewMapTask(taskTracker, numTaskTrackers, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; taskTrackerManager.getNumberOfUniqueHosts()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (t != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return Collections.singletonList(t); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; //分配完map task，再分配reduce task &nbsp; if (numReduces &lt; maxReduceLoad) { &nbsp;&nbsp;&nbsp; int totalNeededReduces = 0; &nbsp;&nbsp;&nbsp; synchronized (jobQueue) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (JobInProgress job : jobQueue) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (job.getStatus().getRunState() != JobStatus.RUNNING || &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job.numReduceTasks == 0) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; continue; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Task t = job.obtainNewReduceTask(taskTracker, numTaskTrackers, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; taskTrackerManager.getNumberOfUniqueHosts()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (t != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return Collections.singletonList(t); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; return null; } </td> 
 </tr> 
 </tbody> 
</table>

从上面的代码中我们可以知道，JobInProgress的obtainNewMapTask是用来分配map task的，其主要调用findNewMapTask，根据TaskTracker所在的Node从nonRunningMapCache中查找TaskInProgress。JobInProgress的obtainNewReduceTask是用来分配reduce task的，其主要调用findNewReduceTask，从nonRunningReduces查找TaskInProgress。

# 五、TaskTracker #

在向JobTracker发送heartbeat后，返回的reponse中有分配好的任务LaunchTaskAction，将其加入队列，调用addToTaskQueue，如果是map task则放入mapLancher(类型为TaskLauncher)，如果是reduce task则放入reduceLancher(类型为TaskLauncher)：

<table> 
 <tbody> 
 <tr> 
 <td> private void addToTaskQueue(LaunchTaskAction action) { &nbsp; if (action.getTask().isMapTask()) { &nbsp;&nbsp;&nbsp; mapLauncher.addToTaskQueue(action); &nbsp; } else { &nbsp;&nbsp;&nbsp; reduceLauncher.addToTaskQueue(action); &nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

TaskLauncher是一个线程，其run函数从上面放入的queue中取出一个TaskInProgress，然后调用startNewTask(TaskInProgress tip)来启动一个task，其又主要调用了localizeJob(TaskInProgress tip)：

<table> 
 <tbody> 
 <tr> 
 <td> private void localizeJob(TaskInProgress tip) throws IOException { &nbsp; //首先要做的一件事情是有关Task的文件从HDFS拷贝的TaskTracker的本地文件系统中：job.split，job.xml以及job.jar &nbsp; Path localJarFile = null; &nbsp; Task t = tip.getTask(); &nbsp; JobID jobId = t.getJobID(); &nbsp; Path jobFile = new Path(t.getJobFile()); &nbsp; …… &nbsp; Path localJobFile = lDirAlloc.getLocalPathForWrite( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; getLocalJobDir(jobId.toString()) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + Path.SEPARATOR + "job.xml", &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobFileSize, fConf); &nbsp; RunningJob rjob = addTaskToJob(jobId, tip); &nbsp; synchronized (rjob) { &nbsp;&nbsp;&nbsp; if (!rjob.localized) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; FileSystem localFs = FileSystem.getLocal(fConf); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Path jobDir = localJobFile.getParent(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //将job.split拷贝到本地 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; systemFS.copyToLocalFile(jobFile, localJobFile); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; JobConf localJobConf = new JobConf(localJobFile); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Path workDir = lDirAlloc.getLocalPathForWrite( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (getLocalJobDir(jobId.toString()) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + Path.SEPARATOR + "work"), fConf); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (!localFs.mkdirs(workDir)) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new IOException("Mkdirs failed to create " &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + workDir.toString()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; System.setProperty("job.local.dir", workDir.toString()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; localJobConf.set("job.local.dir", workDir.toString()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // copy Jar file to the local FS and unjar it. &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String jarFile = localJobConf.getJar(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long jarFileSize = -1; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (jarFile != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Path jarFilePath = new Path(jarFile); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; localJarFile = new Path(lDirAlloc.getLocalPathForWrite( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; getLocalJobDir(jobId.toString()) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + Path.SEPARATOR + "jars", &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 5 * jarFileSize, fConf), "job.jar"); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (!localFs.mkdirs(localJarFile.getParent())) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new IOException("Mkdirs failed to create jars directory "); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //将job.jar拷贝到本地 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; systemFS.copyToLocalFile(jarFilePath, localJarFile); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; localJobConf.setJar(localJarFile.toString()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //将job得configuration写成job.xml &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; OutputStream out = localFs.create(localJobFile); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; localJobConf.writeXml(out); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } finally { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.close(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 解压缩job.jar &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; RunJar.unJar(new File(localJarFile.toString()), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new File(localJarFile.getParent().toString())); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; rjob.localized = true; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; rjob.jobConf = localJobConf; &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; //真正的启动此Task &nbsp; launchTaskForJob(tip, new JobConf(rjob.jobConf)); } </td> 
 </tr> 
 </tbody> 
</table>

当所有的task运行所需要的资源都拷贝到本地后，则调用launchTaskForJob，其又调用TaskInProgress的launchTask函数：

<table> 
 <tbody> 
 <tr> 
 <td> public synchronized void launchTask() throws IOException { &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; //创建task运行目录 &nbsp;&nbsp;&nbsp; localizeTask(task); &nbsp;&nbsp;&nbsp; if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; this.taskStatus.setRunState(TaskStatus.State.RUNNING); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; //创建并启动TaskRunner，对于MapTask，创建的是MapTaskRunner，对于ReduceTask，创建的是ReduceTaskRunner &nbsp;&nbsp;&nbsp; this.runner = task.createRunner(TaskTracker.this, this); &nbsp;&nbsp;&nbsp; this.runner.start(); &nbsp;&nbsp;&nbsp; this.taskStatus.setStartTime(System.currentTimeMillis()); } </td> 
 </tr> 
 </tbody> 
</table>

TaskRunner是一个线程，其run函数如下：

<table> 
 <tbody> 
 <tr> 
 <td> public final void run() { &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; TaskAttemptID taskid = t.getTaskID(); &nbsp;&nbsp;&nbsp; LocalDirAllocator lDirAlloc = new LocalDirAllocator("mapred.local.dir"); &nbsp;&nbsp;&nbsp; File jobCacheDir = null; &nbsp;&nbsp;&nbsp; if (conf.getJar() != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jobCacheDir = new File( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Path(conf.getJar()).getParent().toString()); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; File workDir = new File(lDirAlloc.getLocalPathToRead( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; TaskTracker.getLocalTaskDir( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; t.getJobID().toString(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; t.getTaskID().toString(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; t.isTaskCleanupTask()) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + Path.SEPARATOR + MRConstants.WORKDIR, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; conf). toString()); &nbsp;&nbsp;&nbsp; FileSystem fileSystem; &nbsp;&nbsp;&nbsp; Path localPath; &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; //拼写classpath &nbsp;&nbsp;&nbsp; String baseDir; &nbsp;&nbsp;&nbsp; String sep = System.getProperty("path.separator"); &nbsp;&nbsp;&nbsp; StringBuffer classPath = new StringBuffer(); &nbsp;&nbsp;&nbsp; // start with same classpath as parent process &nbsp;&nbsp;&nbsp; classPath.append(System.getProperty("java.class.path")); &nbsp;&nbsp;&nbsp; classPath.append(sep); &nbsp;&nbsp;&nbsp; if (!workDir.mkdirs()) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (!workDir.isDirectory()) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; LOG.fatal("Mkdirs failed to create " + workDir.toString()); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; String jar = conf.getJar(); &nbsp;&nbsp;&nbsp; if (jar != null) {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // if jar exists, it into workDir &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; File[] libs = new File(jobCacheDir, "lib").listFiles(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (libs != null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (int i = 0; i &lt; libs.length; i++) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(sep);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // add libs from jar to classpath &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(libs[i]); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(sep); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(new File(jobCacheDir, "classes")); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(sep); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; classPath.append(jobCacheDir); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; classPath.append(sep); &nbsp;&nbsp;&nbsp; classPath.append(workDir); &nbsp;&nbsp;&nbsp; //拼写命令行java及其参数 &nbsp;&nbsp;&nbsp; Vector&lt;String&gt; vargs = new Vector&lt;String&gt;(8); &nbsp;&nbsp;&nbsp; File jvm = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new File(new File(System.getProperty("java.home"), "bin"), "java"); &nbsp;&nbsp;&nbsp; vargs.add(jvm.toString()); &nbsp;&nbsp;&nbsp; String javaOpts = conf.get("mapred.child.java.opts", "-Xmx200m"); &nbsp;&nbsp;&nbsp; javaOpts = javaOpts.replace("@taskid@", taskid.toString()); &nbsp;&nbsp;&nbsp; String [] javaOptsSplit = javaOpts.split(" "); &nbsp;&nbsp;&nbsp; String libraryPath = System.getProperty("java.library.path"); &nbsp;&nbsp;&nbsp; if (libraryPath == null) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; libraryPath = workDir.getAbsolutePath(); &nbsp;&nbsp;&nbsp; } else { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; libraryPath += sep + workDir; &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; boolean hasUserLDPath = false; &nbsp;&nbsp;&nbsp; for(int i=0; i&lt;javaOptsSplit.length ;i++) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if(javaOptsSplit[i].startsWith("-Djava.library.path=")) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; javaOptsSplit[i] += sep + libraryPath; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; hasUserLDPath = true; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; break; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; if(!hasUserLDPath) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; vargs.add("-Djava.library.path=" + libraryPath); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; for (int i = 0; i &lt; javaOptsSplit.length; i++) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; vargs.add(javaOptsSplit[i]); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; //添加Child进程的临时文件夹 &nbsp;&nbsp;&nbsp; String tmp = conf.get("mapred.child.tmp", "./tmp"); &nbsp;&nbsp;&nbsp; Path tmpDir = new Path(tmp); &nbsp;&nbsp;&nbsp; if (!tmpDir.isAbsolute()) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tmpDir = new Path(workDir.toString(), tmp); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; FileSystem localFs = FileSystem.getLocal(conf); &nbsp;&nbsp;&nbsp; if (!localFs.mkdirs(tmpDir) &amp;&amp; !localFs.getFileStatus(tmpDir).isDir()) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new IOException("Mkdirs failed to create " + tmpDir.toString()); &nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; vargs.add("-Djava.io.tmpdir=" + tmpDir.toString()); &nbsp;&nbsp;&nbsp; // Add classpath. &nbsp;&nbsp;&nbsp; vargs.add("-classpath"); &nbsp;&nbsp;&nbsp; vargs.add(classPath.toString()); &nbsp;&nbsp;&nbsp; //log文件夹 &nbsp;&nbsp;&nbsp; long logSize = TaskLog.getTaskLogLength(conf); &nbsp;&nbsp;&nbsp; vargs.add("-Dhadoop.log.dir=" + &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new File(System.getProperty("hadoop.log.dir") &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ).getAbsolutePath()); &nbsp;&nbsp;&nbsp; vargs.add("-Dhadoop.root.logger=INFO,TLA"); &nbsp;&nbsp;&nbsp; vargs.add("-Dhadoop.tasklog.taskid=" + taskid); &nbsp;&nbsp;&nbsp; vargs.add("-Dhadoop.tasklog.totalLogFileSize=" + logSize); &nbsp;&nbsp;&nbsp; // 运行map task和reduce task的子进程的main class是Child &nbsp;&nbsp;&nbsp; vargs.add(Child.class.getName());&nbsp; // main of Child &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; //运行子进程 &nbsp;&nbsp;&nbsp; jvmManager.launchJvm(this, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jvmManager.constructJvmEnv(setup,vargs,stdout,stderr,logSize, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; workDir, env, pidFile, conf)); } </td> 
 </tr> 
 </tbody> 
</table>

# 六、Child #

真正的map task和reduce task都是在Child进程中运行的，Child的main函数的主要逻辑如下：

<table> 
 <tbody> 
 <tr> 
 <td> while (true) { &nbsp; //从TaskTracker通过网络通信得到JvmTask对象 &nbsp; JvmTask myTask = umbilical.getTask(jvmId); &nbsp; …… &nbsp; idleLoopCount = 0; &nbsp; task = myTask.getTask(); &nbsp; taskid = task.getTaskID(); &nbsp; isCleanup = task.isTaskCleanupTask(); &nbsp; JobConf job = new JobConf(task.getJobFile()); &nbsp; TaskRunner.setupWorkDir(job); &nbsp; numTasksToExecute = job.getNumTasksToExecutePerJvm(); &nbsp; task.setConf(job); &nbsp; defaultConf.addResource(new Path(task.getJobFile())); &nbsp; …… &nbsp; //运行task &nbsp; task.run(job, umbilical);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // run the task &nbsp; if (numTasksToExecute &gt; 0 &amp;&amp; ++numTasksExecuted == numTasksToExecute) { &nbsp;&nbsp;&nbsp; break; &nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

## 6.1、MapTask ##

如果task是MapTask，则其run函数如下：

<table> 
 <tbody> 
 <tr> 
 <td> public void run(final JobConf job, final TaskUmbilicalProtocol umbilical) &nbsp; throws IOException { &nbsp; //用于同TaskTracker进行通信，汇报运行状况 &nbsp; final Reporter reporter = getReporter(umbilical); &nbsp; startCommunicationThread(umbilical); &nbsp; initialize(job, reporter); &nbsp; …… &nbsp; //map task的输出 &nbsp; int numReduceTasks = conf.getNumReduceTasks(); &nbsp; MapOutputCollector collector = null; &nbsp; if (numReduceTasks &gt; 0) { &nbsp;&nbsp;&nbsp; collector = new MapOutputBuffer(umbilical, job, reporter); &nbsp; } else { &nbsp;&nbsp;&nbsp; collector = new DirectMapOutputCollector(umbilical, job, reporter); &nbsp; } &nbsp; //读取input split，按照其中的信息，生成RecordReader来读取数据 instantiatedSplit = (InputSplit) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ReflectionUtils.newInstance(job.getClassByName(splitClass), job); &nbsp; DataInputBuffer splitBuffer = new DataInputBuffer(); &nbsp; splitBuffer.reset(split.getBytes(), 0, split.getLength()); &nbsp; instantiatedSplit.readFields(splitBuffer); &nbsp; if (instantiatedSplit instanceof FileSplit) { &nbsp;&nbsp;&nbsp; FileSplit fileSplit = (FileSplit) instantiatedSplit; &nbsp;&nbsp;&nbsp; job.set("map.input.file", fileSplit.getPath().toString()); &nbsp;&nbsp;&nbsp; job.setLong("map.input.start", fileSplit.getStart()); &nbsp;&nbsp;&nbsp; job.setLong("map.input.length", fileSplit.getLength()); &nbsp; } &nbsp; RecordReader rawIn =&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // open input &nbsp;&nbsp;&nbsp; job.getInputFormat().getRecordReader(instantiatedSplit, job, reporter); &nbsp; RecordReader in = isSkipping() ? &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new SkippingRecordReader(rawIn, getCounters(), umbilical) : &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new TrackedRecordReader(rawIn, getCounters()); &nbsp; job.setBoolean("mapred.skip.on", isSkipping()); &nbsp; //对于map task，生成一个MapRunnable，默认是MapRunner &nbsp; MapRunnable runner = &nbsp;&nbsp;&nbsp; ReflectionUtils.newInstance(job.getMapRunnerClass(), job); &nbsp; try { &nbsp;&nbsp;&nbsp; //MapRunner的run函数就是依次读取RecordReader中的数据，然后调用Mapper的map函数进行处理。 &nbsp;&nbsp;&nbsp; runner.run(in, collector, reporter);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp; collector.flush(); &nbsp; } finally { &nbsp;&nbsp;&nbsp; in.close();&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // close input &nbsp;&nbsp;&nbsp; collector.close(); &nbsp; } &nbsp; done(umbilical); } </td> 
 </tr> 
 </tbody> 
</table>

MapRunner的run函数就是依次读取RecordReader中的数据，然后调用Mapper的map函数进行处理：

<table> 
 <tbody> 
 <tr> 
 <td> public void run(RecordReader&lt;K1, V1&gt; input, OutputCollector&lt;K2, V2&gt; output, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Reporter reporter) &nbsp; throws IOException { &nbsp; try { &nbsp;&nbsp;&nbsp; K1 key = input.createKey(); &nbsp;&nbsp;&nbsp; V1 value = input.createValue(); &nbsp;&nbsp;&nbsp; while (input.next(key, value)) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mapper.map(key, value, output, reporter); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if(incrProcCount) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; reporter.incrCounter(SkipBadRecords.COUNTER_GROUP, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; SkipBadRecords.COUNTER_MAP_PROCESSED_RECORDS, 1); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } &nbsp; } finally { &nbsp;&nbsp;&nbsp; mapper.close(); &nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

结果集全部收集到MapOutputBuffer中，其collect函数如下：

<table> 
 <tbody> 
 <tr> 
 <td> public synchronized void collect(K key, V value) &nbsp;&nbsp;&nbsp; throws IOException { &nbsp; reporter.progress(); &nbsp; …… &nbsp; //从此处看，此buffer是一个ring的数据结构 &nbsp; final int kvnext = (kvindex + 1) % kvoffsets.length; &nbsp; spillLock.lock(); &nbsp; try { &nbsp;&nbsp;&nbsp; boolean kvfull; &nbsp;&nbsp;&nbsp; do { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //在ring中，如果下一个空闲位置接上起始位置的话，则表示满了 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; kvfull = kvnext == kvstart; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //在ring中计算是否需要将buffer写入硬盘的阈值 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; final boolean kvsoftlimit = ((kvnext &gt; kvend) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ? kvnext - kvend &gt; softRecordLimit &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; : kvend - kvnext &lt;= kvoffsets.length - softRecordLimit); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //如果到达阈值，则开始将buffer写入硬盘，写成spill文件。 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //startSpill主要是notify一个背后线程SpillThread的run()函数，开始调用sortAndSpill()开始排序，合并，写入硬盘 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (kvstart == kvend &amp;&amp; kvsoftlimit) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; startSpill(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //如果buffer满了，则只能等待写入完毕 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (kvfull) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while (kvstart != kvend) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; reporter.progress(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; spillDone.await(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp; } while (kvfull); &nbsp; } finally { &nbsp;&nbsp;&nbsp; spillLock.unlock(); &nbsp; } &nbsp; try { &nbsp;&nbsp;&nbsp; //如果buffer不满，则将key, value写入buffer &nbsp;&nbsp;&nbsp; int keystart = bufindex; &nbsp;&nbsp;&nbsp; keySerializer.serialize(key); &nbsp;&nbsp;&nbsp; final int valstart = bufindex; &nbsp;&nbsp;&nbsp; valSerializer.serialize(value); &nbsp;&nbsp;&nbsp; int valend = bb.markRecord(); &nbsp;&nbsp;&nbsp; //调用设定的partitioner，根据key, value取得partition id &nbsp;&nbsp;&nbsp; final int partition = partitioner.getPartition(key, value, partitions); &nbsp;&nbsp;&nbsp; mapOutputRecordCounter.increment(1); &nbsp;&nbsp;&nbsp; mapOutputByteCounter.increment(valend &gt;= keystart &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ? valend - keystart &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; : (bufvoid - keystart) + valend); &nbsp;&nbsp;&nbsp; //将parition id以及key, value在buffer中的偏移量写入索引数组 &nbsp;&nbsp;&nbsp; int ind = kvindex * ACCTSIZE; &nbsp;&nbsp;&nbsp; kvoffsets[kvindex] = ind; &nbsp;&nbsp;&nbsp; kvindices[ind + PARTITION] = partition; &nbsp;&nbsp;&nbsp; kvindices[ind + KEYSTART] = keystart; &nbsp;&nbsp;&nbsp; kvindices[ind + VALSTART] = valstart; &nbsp;&nbsp;&nbsp; kvindex = kvnext; &nbsp; } catch (MapBufferTooSmallException e) { &nbsp;&nbsp;&nbsp; LOG.info("Record too large for in-memory buffer: " + e.getMessage()); &nbsp;&nbsp;&nbsp; spillSingleRecord(key, value); &nbsp;&nbsp;&nbsp; mapOutputRecordCounter.increment(1); &nbsp;&nbsp;&nbsp; return; &nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

内存buffer的格式如下：

(见几位hadoop大侠的分析[http://blog.csdn.net/HEYUTAO007/archive/2010/07/10/5725379.aspx][http_blog.csdn.net_HEYUTAO007_archive_2010_07_10_5725379.aspx] 以及[http://caibinbupt.javaeye.com/][http_caibinbupt.javaeye.com])

[![clip\_image001][clip_image001]][clip_image001_clip_image001]

kvoffsets是为了写入内存前排序使用的。

从上面可知，内存buffer写入硬盘spill文件的函数为sortAndSpill：

<table> 
 <tbody> 
 <tr> 
 <td> private void sortAndSpill() throws IOException { &nbsp; …… &nbsp; FSDataOutputStream out = null; &nbsp; FSDataOutputStream indexOut = null; &nbsp; IFileOutputStream indexChecksumOut = null; &nbsp; //创建硬盘上的spill文件 &nbsp; Path filename = mapOutputFile.getSpillFileForWrite(getTaskID(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numSpills, size); &nbsp; out = rfs.create(filename); &nbsp; …… &nbsp; final int endPosition = (kvend &gt; kvstart) &nbsp;&nbsp;&nbsp; ? kvend &nbsp;&nbsp;&nbsp; : kvoffsets.length + kvend; &nbsp; //按照partition的顺序对buffer中的数据进行排序 &nbsp; sorter.sort(MapOutputBuffer.this, kvstart, endPosition, reporter); &nbsp; int spindex = kvstart; &nbsp; InMemValBytes value = new InMemValBytes(); &nbsp; //依次一个一个parition的写入文件 &nbsp; for (int i = 0; i &lt; partitions; ++i) { &nbsp;&nbsp;&nbsp; IFile.Writer&lt;K, V&gt; writer = null; &nbsp;&nbsp;&nbsp; long segmentStart = out.getPos(); &nbsp;&nbsp;&nbsp; writer = new Writer&lt;K, V&gt;(job, out, keyClass, valClass, codec); &nbsp;&nbsp;&nbsp; //如果combiner为空，则直接写入文件 &nbsp;&nbsp;&nbsp; if (null == combinerClass) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; writer.append(key, value); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ++spindex; &nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp; else { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //如果combiner不为空，则先combine，调用combiner.reduce(…)函数后再写入文件 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; combineAndSpill(kvIter, combineInputCounter); &nbsp;&nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; …… } </td> 
 </tr> 
 </tbody> 
</table>

当map阶段结束的时候，MapOutputBuffer的flush函数会被调用，其也会调用sortAndSpill将buffer中的写入文件，然后再调用mergeParts来合并写入在硬盘上的多个spill:

<table> 
 <tbody> 
 <tr> 
 <td> private void mergeParts() throws IOException { &nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; //对于每一个partition &nbsp;&nbsp;&nbsp; for (int parts = 0; parts &lt; partitions; parts++){ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //create the segments to be merged &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; List&lt;Segment&lt;K, V&gt;&gt; segmentList = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new ArrayList&lt;Segment&lt;K, V&gt;&gt;(numSpills); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; TaskAttemptID mapId = getTaskID(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //依次从各个spill文件中收集属于当前partition的段 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for(int i = 0; i &lt; numSpills; i++) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; final IndexRecord indexRecord = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; getIndexInformation(mapId, i, parts); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long segmentOffset = indexRecord.startOffset; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long segmentLength = indexRecord.partLength; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Segment&lt;K, V&gt; s = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Segment&lt;K, V&gt;(job, rfs, filename[i], segmentOffset, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; segmentLength, codec, true); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; segmentList.add(i, s); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //将属于同一个partition的段merge到一起 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; RawKeyValueIterator kvIter = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Merger.merge(job, rfs, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; keyClass, valClass, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; segmentList, job.getInt("io.sort.factor", 100), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Path(getTaskID().toString()), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job.getOutputKeyComparator(), reporter); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //写入合并后的段到文件 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long segmentStart = finalOut.getPos(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Writer&lt;K, V&gt; writer = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Writer&lt;K, V&gt;(job, finalOut, keyClass, valClass, codec); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (null == combinerClass || numSpills &lt; minSpillsForCombine) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Merger.writeFile(kvIter, writer, reporter, job); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } else { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; combineCollector.setWriter(writer); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; combineAndSpill(kvIter, combineInputCounter); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; } } </td> 
 </tr> 
 </tbody> 
</table>

## 6.2、ReduceTask ##

ReduceTask的run函数如下：

<table> 
 <tbody> 
 <tr> 
 <td> public void run(JobConf job, final TaskUmbilicalProtocol umbilical) &nbsp; throws IOException { &nbsp; job.setBoolean("mapred.skip.on", isSkipping()); &nbsp; //对于reduce，则包含三个步骤：拷贝，排序，Reduce &nbsp; if (isMapOrReduce()) { &nbsp;&nbsp;&nbsp; copyPhase = getProgress().addPhase("copy"); &nbsp;&nbsp;&nbsp; sortPhase&nbsp; = getProgress().addPhase("sort"); &nbsp;&nbsp;&nbsp; reducePhase = getProgress().addPhase("reduce"); &nbsp; } &nbsp; startCommunicationThread(umbilical); &nbsp; final Reporter reporter = getReporter(umbilical); &nbsp; initialize(job, reporter); &nbsp; //copy阶段，主要使用ReduceCopier的fetchOutputs函数获得map的输出。创建多个线程MapOutputCopier，其中copyOutput进行拷贝。 &nbsp; boolean isLocal = "local".equals(job.get("mapred.job.tracker", "local")); &nbsp; if (!isLocal) { &nbsp;&nbsp;&nbsp; reduceCopier = new ReduceCopier(umbilical, job); &nbsp;&nbsp;&nbsp; if (!reduceCopier.fetchOutputs()) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; …… &nbsp;&nbsp;&nbsp; } &nbsp; } &nbsp; copyPhase.complete(); &nbsp; //sort阶段，将得到的map输出合并，直到文件数小于io.sort.factor时停止，返回一个Iterator用于访问key-value &nbsp; setPhase(TaskStatus.Phase.SORT); &nbsp; statusUpdate(umbilical); &nbsp; final FileSystem rfs = FileSystem.getLocal(job).getRaw(); &nbsp; RawKeyValueIterator rIter = isLocal &nbsp;&nbsp;&nbsp; ? Merger.merge(job, rfs, job.getMapOutputKeyClass(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job.getMapOutputValueClass(), codec, getMapFiles(rfs, true), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; !conf.getKeepFailedTaskFiles(), job.getInt("io.sort.factor", 100), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Path(getTaskID().toString()), job.getOutputKeyComparator(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; reporter) &nbsp;&nbsp;&nbsp; : reduceCopier.createKVIterator(job, rfs, reporter); &nbsp; mapOutputFilesOnDisk.clear(); &nbsp; sortPhase.complete(); &nbsp; //reduce阶段 &nbsp; setPhase(TaskStatus.Phase.REDUCE); &nbsp; …… &nbsp; Reducer reducer = ReflectionUtils.newInstance(job.getReducerClass(), job); &nbsp; Class keyClass = job.getMapOutputKeyClass(); &nbsp; Class valClass = job.getMapOutputValueClass(); &nbsp; ReduceValuesIterator values = isSkipping() ? &nbsp;&nbsp;&nbsp;&nbsp; new SkippingReduceValuesIterator(rIter, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job.getOutputValueGroupingComparator(), keyClass, valClass, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job, reporter, umbilical) : &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new ReduceValuesIterator(rIter, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job.getOutputValueGroupingComparator(), keyClass, valClass, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; job, reporter); &nbsp; //逐个读出key-value list，然后调用Reducer的reduce函数 &nbsp; while (values.more()) { &nbsp;&nbsp;&nbsp; reduceInputKeyCounter.increment(1); &nbsp;&nbsp;&nbsp; reducer.reduce(values.getKey(), values, collector, reporter); &nbsp;&nbsp;&nbsp; values.nextKey(); &nbsp;&nbsp;&nbsp; values.informReduceProgress(); &nbsp; } &nbsp; reducer.close(); &nbsp; out.close(reporter); &nbsp; done(umbilical); } </td> 
 </tr> 
 </tbody> 
</table>

# 七、总结 #

Map-Reduce的过程总结如下图：

[![map reduce][]][map reduce 1]

[http_blog.csdn.net_HEYUTAO007_archive_2010_07_10_5725379.aspx]: http://blog.csdn.net/HEYUTAO007/archive/2010/07/10/5725379.aspx
[http_caibinbupt.javaeye.com]: http://caibinbupt.javaeye.com/
[clip_image001]: https://images.cnblogs.com/cnblogs_com/forfuture1978/WindowsLiveWriter/HadoopMapReduce_14A45/clip_image001_thumb.jpg
[clip_image001_clip_image001]: /images/20220329/27c3f9ce3cd04b37bcef4a741212fe20.png
[map reduce]: /images/20220329/241191067711416796ffc9cb5b90c461.png
[map reduce 1]: http://images.cnblogs.com/cnblogs_com/forfuture1978/WindowsLiveWriter/HadoopMapReduce_14A45/map%20reduce_2.jpg