MapReduce流程分析

MapReduce流程分析

接触Hadoop已经1年了，一直没时间好好学习下。这几天打算好好研究下Hadoop.本来是想打算改写下TextInputFormat。看了源码后，反而更迷糊了。所以干脆连MapReduce的整个流程写下来。也当为这几天的学习作个总结。

先来一个我们常写的main函数。

Configuration conf = new Configuration();

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

if (otherArgs.length != 2){

System.exit(2);

}

Job job = new Job(conf, "wordcount");

 

job.setJarByClass(mywordcount.class);

 

job.setInputFormatClass(TextInputFormat.class);

 

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

 

job.setMapperClass(wordcountMapper.class);

job.setReducerClass(wordcountReduce.class);

job.setCombinerClass(wordcountReduce.class);

 

FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));

FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

 

job.waitForCompletion(true);

上述程序我就不分析。直接来分析下运行流程。

在master节点的NameNode, SecondedNameNode,JobTracker和slaves节点的DataNode, TaskTracker都已经启动后，JobTracker一直在等待JobClient通过RPC提交作业,而TaskTracker一直通过RPC向 JobTracker发送心跳heartbeat询问有没有任务可做。

而主程序中通过job.waitForCompletion(true)函数通过调用JobClient.runJob()函数将MapReduce作业交与JobTrack进行执行。

1：JobClient.runJob()

runjob会根据用户的设置（job.setInputFormatClass（））来将需要输入的数据划分成小的数据集，同时返回划分后split想的相应信息（这里的split路径信息我估计应该是hdfs里的路径和偏移，因为我们要处理的数据早应该上传到了hdfs系统）。同时根据split设置MapTask的个数。获取了上面信息后，就将运行任务所需要的全部数据、信息全部上传至HDFS。上传的内容主要包括三个包： job.jar, job.split和job.xml

job.xml: 作业配置，例如Mapper, Combiner, Reducer的类型，输入输出格式的类型等。
job.jar: jar包,里面包含了执行此任务需要的各种类，比如 Mapper,Reducer等实现。
job.split: 文件分块的相关信息，比如有数据分多少个块，块的大小(默认64m)等。

进行完上述工作后，本地的工作就完成了，函数jobSubmitClient.submitJob(jobId)调用真正的JobTracker执行Task。

2:JobTracker.submitJob()

当JobTracker接收到新的job请求（即submitJob()函数被调用）后，会创建一个JobInProgress对象并通过它来管理和调度任务。JobInProgress在创建的时候会初始化一系列与任务有关的参数，调用FileSystem，把在JobClient端上传的所有任务文件下载到本地的文件系统中的临时目录里。这其中包括上传的*.jar文件包、记录配置信息的xml、记录分割信息的文件。

在JobTracker的构造函数中，会生成一个taskScheduler成员变量，来进行Job的调度,默认为JobQueueTaskScheduler，也即按照FIFO的方式调度任务。而在offerService函数中也为JobTracker(也即taskScheduler的taskTrackerManager)注册了两个Listener：

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

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

3：initTasks（）

任务Task分两种: MapTask 和reduceTask，它们的管理对象都是TaskInProgress 。

在initTasks函数，会通过JobClient的readSplitFile()获得已分解的输入数据的RawSplit列表，然后通过这个列表创建相应的TaskInProgress(MapTask)同时还会读取相应数据块所在DataNode的主机名（通过FileSplit的getLocations()函数获取）。创建完TaskInProgress后，就会调用createCache()方法为这些TaskInProgress对象产生一个未执行任务的Map缓存nonRunningMapCache。当TaskTracker向JobTrack中发送心跳，请求任务时，就会去直接去这个缓存中取任务。

JobInProgress也会创建Reduce的监控对象- TaskInProgress，而这个的数量就是根据用户在程序里的设置，默认的是一个。同样地，initTasks()也会通过createCache()方法产生nonRunningReduceCache成员。JobInProgress再接着进行清理工作。最后再记录一下Job的执行日志。

至此Job的初始化就全部完成。

4：TaskTracker

TaskTracker从启动后，就每隔一定的时间向JobTracker发送一次心跳(默认10s，发送的内容包括自己的当前状态，当满足一定状态时就可以向JobTracker申请新的任务。如Map Task、 Reduce Task都还有运行的能力)通过transmitHeartBeat()发送心跳后再接受JobTracker返回的HeartbeatResponse。然后调用HeartbeatResponse的getActions()函数获得JobTracker传过来的所有指令即一个TaskTrackerAction数组。再遍历这个数组，就可以知道需要完成的事情。（这些事情可能是LaunchTaskAction 执行新任务、 KillTaskAction 结束一个任务）如果有分配好的任务将其加入队列，调用addToTaskQueue，如果是map task则放入mapLancher(类型为TaskLauncher)，如果是reduce task则放入reduceLancher(类型为TaskLauncher)

5 JobTracker：heartbeat（）

JobTracker是通过heartbeat（）函数来接受TaskTracker的心跳，如果TaskTracker是请求任务的指令。Heartbeat（）函数就会调用默认的任务调度器（JobQueueTaskScheduler）来分配任务。先计算Map和Reduce的剩余工作量，再计算每个TaskTracker应有的工作量。如果TaskTracker上运行的map task数目小于平均的工作量，则向其分配map task。分配完Map Task后再分配Reduce task.而这里有一个函数findNewMapTask（）就是从nonRunningMapCache和nonRunningReduceCache中查找出map task的TaskInProgress和Reduce task. 的TaskInProgress再返回给TaskTracker。

findNewMapTask()从近到远一层一层地寻找，首先是同一节点，然后在寻找同一机柜上的节点，接着寻找相同数据中心下的节点，直到找了maxLevel层结束。这样的话，在JobTracker给TaskTracker派发任务的时候，可以迅速找到最近的TaskTracker，让它执行任务。(通过寻找本任务split所在的DataNode,然后判断发送心跳的TaskTracker和本任务split所在的DataNode是不是同一主机。如果是则分配这个MapTask给发送心跳的Tracker，如果不是者返回null不进行分配。)

再调用localizeJob（）进行真正的初始化（TaskTracker上的Task）。而localizeJob又调用TaskLauncher

6：TaskLauncher

TaskLauncher是一个线程就是从上面的队列中取出TaskInProgress然后调用startNewTask(TaskInProgress tip)来启动一个task。

这里又会再次将Task有关的数据包、信息包从HDFS拷贝回本地文件系统包括：job.split，job.xml以及job.jar，当所有的资源拷贝回来后，就调用launchTaskForJob()开始执行Task.

launchTaskForJob函数又调用launchTask（）

7：launchTask（）

launchTask()函数首先通过createRunner（）函数是创建MapTaskRunner来启动子进程和创建ReduceTaskRunner来启动子进程。TaskRunner负责将一个任务放到一个进程里面来执行。它会调用run()函数来处理。run()函数会初始化一系列环境变量等。最后生成一个新进程并运行即runChild。

8 Child进程

真正的map task和reduce task都是在Child进程中运行的。Child进程会运行Task.

9：MapTask

如果是MapTask，MapTask.run()首先向TaskTracker汇报情况，再设置Mapper的输出格式。接着读取input split，按照其中的信息，生成RecordReader来读取数据。这其中会生成一个MapRunnable

，而MapRunnable要完成的任务就时通过RecordReader的next函数读取循环从split中读取<k,v>交给map函数进行处理，然后使用OutputCollector收集每次处理<k,v>对后得到的新的<k,v>对.

10:OutputCollector

OutputCollector的作用是收集每次调用map后得到的新的kv对，宁把他们spill到文件或者放到内存，以做进一步的处理，比如排序，combine等。

MapOutputCollector 有两个子类：MapOutputBuffer和DirectMapOutputCollector。 DirectMapOutputCollector用在不需要Reduce阶段的时候。如果Mapper后续有reduce任务，系统会使用MapOutputBuffer做为输出， MapOutputBuffer使用了一个缓冲区对map的处理结果进行缓存，放在内存中.在适当的时机，缓冲区中的数据会被spill到硬盘中。spillThread线程实现将缓冲区的数据写入硬盘。

向硬盘中写数据的时机:

（1）当内存缓冲区不能容下一个太大的kv对时。spillSingleRecord方法。

（2）内存缓冲区已满时。SpillThread线程。

（3）Mapper的结果都已经collect了，需要对缓冲区做最后的清理。Flush方法。

11：ReduceTask

ReduceTask .run()函数同样先进行一系列的初始化工作。之后进入正式的工作，主要有这么三个步骤：Copy、Sort、Reduce。

11.1:copy

copy就是从执行各个Map任务的服务器那里，搜罗map的输出文件。

拷贝的任务的是由ReduceTask.ReduceCopier 类来负责。ReduceCopier先向父TaskTracker询问此作业个Map任务的完成状况，获取到map服务器的相关信息后由线程MapOutputCopier做具体的拷贝工作。在拷贝过来的同时也会做一些归并排序以减轻后面sort的负担。

11.2 Sort

排序工作，就相当于上述排序工作的一个延续。它会在所有的文件都拷贝完毕后进行。使用工具类Merger归并所有的文件。经过这一个流程，一个合并了所有所需Map任务输出文件的新文件产生了。而那些从其他各个服务器网罗过来的 Map任务输出文件，全部删除了。

11.3Reduce

Reduce任务的最后一个阶段。

输入方面：他会准备根据自定义或默认的KeyClass、ValueClass构造出Reducer所需的键类型，和值的迭代类型Iterator

输出方面：它会准备一个OutputCollector收集输出与MapTask不同，这个OutputCollector更为简单，仅仅是打开一个RecordWriter，collect一次(排序完成的那个文件)，write一次（写往HDFS）。

有了输入，有了输出，不断循环调用自定义的Reducer，最终，Reduce阶段完成。

 

写本文之际参看了很多牛人的大作，在这里一并感谢。

 

最后，我还有三个问题没理解，请教高手解答一下。

Des1:当Jobtracker向Tasktracker分配任务时是先判断Tasktracker上运行的Task是否小于平均工作量，小于者向其分配Task。（假如我们这里是MapTask。）然后调用函数obtainNewMapTask()中的findNewMapTask（）来查找nonRunningMapCache中的TaskInProgress。
findNewMapTask（）函数会从近到远一层一层地寻找。首先是同一节点，然后在寻找同一机柜上的节点，接着寻找相同数据中心下的节点，直到找了maxLevel层结束。（这段话是我从网上看到的。没理解这句话的意思。）
Q1:我想请问下， findNewMapTask在这里寻找的是什么？（TaskTracker）我的理解是：通过寻找本任务split所在的DataNode,然后判断发送心跳的TaskTracker和本任务split所在的DataNode是不是同一主机。如果是则分配这个MapTask给发送心跳的Tracker，如果不是者返回null不进行分配。如果我理解错了，请解释下。谢谢了。

Des2:当客户端提交任务后，首先会通过用户设置的InputFormat将文件进行划分。而hadoop默认的TextInputFormat.class.查看源码知道。TextInputFormat是继承的FileInputFormat.并且将isSplitable进行了关闭。所以默认的是不对文件进行划分。
Q2:在运行一个MapReduce程序时，原始数据都会提前上传到HDFS文件系统，大于64M的文件都会被划分存储到多个DataNode。。假如我有一个128M的文件上传到了HDFS，那天文件应该被划分成了2份。那么TextInputFormat在处理输入时不对文件进行划分，在TaskTracker处理文件时，处理的是64M,还是128M呢？

（这个问题已经解决，应该是64M）
Q3：如果是64M是不是会开启两个TaskTracker来处理文件，又因为TextInputFormat对文件不进行划分，所以每个
TaskTracker上只会开启一个Map Task来处理Map任务。最后两个TaskTracker启动两个Reduce生成两个文件？