`

MapReduce流程分析

 
阅读更多

原文:http://blog.csdn.net/jackydai987/article/details/6227365

 

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,JobTrackerslaves节点的DataNode, TaskTracker都已经启动后,JobTracker一直在等待JobClient通过RPC提交作业,TaskTracker一直通过RPC JobTracker发送心跳heartbeat询问有没有任务可做。

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

1JobClient.runJob()

runjob会根据用户的设置(job.setInputFormatClass())来将需要输入的数据划分成小的数据集,同时返回划分后split想的相应信息(这里的split路径信息我估计应该是hdfs里的路径和偏移,因为我们要处理的数据早应该上传到了hdfs系统)。同时根据split设置MapTask的个数。获取了上面信息后,就将运行任务所需要的全部数据、信息全部上传至HDFS。上传的内容主要包括三个包: job.jar, job.splitjob.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(也即taskSchedulertaskTrackerManager)注册了两个Listener

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

EagerTaskInitializationListenerjobAdded函数就是向jobInitQueue中添加一个JobInProgress对象,于是自然触发了此Job的初始化操作。由JobInProgressinitTasks函数完成:

3initTasks()

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

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

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

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

4TaskTracker

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

5 JobTrackerheartbeat()

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

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

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

6TaskLauncher

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

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

launchTaskForJob函数又调用launchTask()

7launchTask()

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

8 Child进程

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

9MapTask

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

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

10:OutputCollector

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

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

向硬盘中写数据的时机:

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

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

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

11ReduceTask

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

11.1:copy

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

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

11.2 Sort

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

11.3Reduce

Reduce任务的最后一个阶段。

输入方面:他会准备根据自定义或默认的KeyClassValueClass构造出Reducer所需的键类型, 和值的迭代类型Iterator

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

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

 

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

 

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

Des1:JobtrackerTasktracker分配任务时是先判断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生成两个文件?

分享到:
评论

相关推荐

    MapReduce 2.0源码分析与编程实

    全书分为10章,系统地介绍了HDFS存储系统,Hadoop的文件I/O系统,MapReduce2.0的框架结构和源码分析,MapReduce2.0的配置与测试,MapReduce2.0运行流程,MapReduce2.0高级程序设计以及相关特性等内容。《MapReduce...

    MapReduce2.0源码分析与实战编程

    全书分为10章,系统地介绍了HDFS存储系统,Hadoop的文件I/O系统,MapReduce 2.0的框架结构和源码分析,MapReduce 2.0的配置与测试,MapReduce 2.0运行流程,MapReduce 2.0高级程序设计以及相关特性等内容。...

    Hadoop MapReduce 入门

    ##### 2.1 MapReduce 流程分析 - **输入文件处理**: - 读取输入文件内容,并将其解析为 key-value 对。 - 每一行数据都会被解析成 key-value 对,并传递给 map 函数处理。 - **map 任务处理**: - 实现自定义逻辑...

    MapReduce详细流程

    MapReduce是一种分布式数据处理模型,广泛应用于大数据的处理和分析领域。它的核心思想是将大任务分解成许多小任务,然后并行处理,最后再将结果进行汇总。在MapReduce模型中,主要包括两个阶段:Map阶段和Reduce...

    5.Hadoop入门进阶课程_第5周_MapReduce原理及操作.pdf

    ##### 2.2 MapReduce流程分析 **2.2.1 Map过程** 1. **输入分片**:每个输入分片会被一个map任务处理,默认情况下以HDFS的一个块大小(通常是64MB)作为分片大小。可以通过调整HDFS配置文件中的相关参数来改变块的...

    基于MapReduce分析的招聘信息大数据可视化系统.zip

    《基于MapReduce分析的招聘信息大数据可视化系统》 在当今数据驱动的时代,招聘信息的大数据...通过优化数据处理流程,提升数据分析的准确性和实时性,企业在人才竞争中能够更好地把握先机,实现人力资源的有效配置。

    MapReduce源码分析

    在本分析中,我们将深入MapReduce的源码,理解其内部机制,以及常用API的工作原理。 **Map阶段** Map阶段是MapReduce处理流程的第一步,它的主要任务是将输入数据分割成多个小块(split),然后对每个块执行用户...

    Hadoop.MapReduce.分析

    ### Hadoop.MapReduce 分析 #### 一、概述 Hadoop.MapReduce 是一种分布式计算模型,主要用于处理大规模数据集。其基本思想源自Google提出的MapReduce论文。本文将深入解析Hadoop.MapReduce的工作原理、核心组件...

    MapReduce分析Youtube数据内含源码以及说明书可以自己运行复现.zip

    在这个"MapReduce分析Youtube数据内含源码以及说明书可以自己运行复现.zip"压缩包中,包含了对YouTube数据进行分析的完整流程。这可能包括用户行为数据、视频信息、评论等。通过MapReduce,我们可以对这些数据进行...

    大数据mapreduce案例

    在这个案例中,我们将深入探讨MapReduce的工作流程,以及如何通过代码来实现大数据的分析。 1. MapReduce的基本概念 MapReduce的核心思想是将大规模数据集分割成小块,然后在多台机器上并行处理这些小块。Map阶段...

    map reduce 源码分析流程

    map reduce的全部执行流程,源码分析视图

    实验项目 MapReduce 编程

    实验内容涵盖了从启动全分布模式的Hadoop集群到编写、运行和分析MapReduce应用程序的全过程。 首先,实验启动了Hadoop集群的所有守护进程,包括NameNode(主节点,负责元数据管理)、DataNode(存储数据的节点)、...

    基于hadoop+MapReduce+Java大数据清洗和分析的基本操作流程.zip

    (3)对于大数据清洗和分析的基本操作流程 清洗不符合规范的数据以及不需要采用的特殊数据、通过字典简化数据格式(如实验第二步中用到的行为地址基础数据)、分析数据中对于生产或研究有意义的数据(如实验第五步...

    hadoop mapreduce编程实战

    * MapReduce 程序的执行流程 * MapReduce 程序的优化方法 MapReduce 项目实践 在实践中,我们可以使用 MapReduce 来解决各种大数据处理问题。以下是一些 MapReduce 项目实践: * WordCount 程序编写及代码分析 * ...

    MapReduce基础.pdf

    - **不适合迭代计算**:MapReduce不支持迭代式的算法,这限制了其在某些复杂数据分析场景中的应用。 - **内存管理**:虽然MapReduce能够处理大规模数据,但在处理特别大的数据集时可能会遇到内存不足的问题。 综上...

Global site tag (gtag.js) - Google Analytics