Map-Reduce的过程解析

爱被打了一巴掌 2022-08-06 09:09 147阅读 0赞

# 一、客户端 #

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

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

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

# 二、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函数，其中有如下的逻辑：

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

# 三、TaskTracker #

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

其中transmitHeartBeat主要逻辑如下：

# 四、JobTracker #

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

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

从上面的代码中我们可以知道，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)：

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

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

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

<table style="border:1px solid silver; border-collapse:collapse; word-break:break-word"> 
 <tbody> 
 <tr> 
 <td style="font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> 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函数的主要逻辑如下：

## 6.1、MapTask ##

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

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

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

内存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：

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

## 6.2、ReduceTask ##

ReduceTask的run函数如下：

# 七、总结 #

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]: http://images.cnblogs.com/cnblogs_com/forfuture1978/WindowsLiveWriter/HadoopMapReduce_14A45/clip_image001_thumb.jpg
[clip_image001_clip_image001]: /images/20220806/7f66bc798e9d450791409338816c51eb.png
[map reduce]: /images/20220806/098f7275e67f40b7b3108923e4ddfe54.png
[map reduce 1]: http://images.cnblogs.com/cnblogs_com/forfuture1978/WindowsLiveWriter/HadoopMapReduce_14A45/map%20reduce_2.jpg