- 浏览: 64259 次
- 性别:
- 来自: 上海
文章分类
最新评论
原文: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,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生成两个文件?
发表评论
-
centos7下安装hadoop2.6.4过程总结
2016-03-03 21:49 0linux是从ubuntu开始接触的,但听说做服务器的,而且 ... -
[转]Yahoo持续的Pig/Hadoop(MapReduce)工作流
2013-03-16 21:40 851传送门:http://blog.sina.c ... -
有关kvm
2013-03-16 21:15 1319有一台式机,希望能模拟一个三四台的小集群,因为装的ubun ... -
关于全加权最小一乘法
2013-02-26 23:40 760这里是网址: http://www.doc88.com/ ... -
[转]包云岗 。普林斯顿 。《世纪图灵纪念》
2012-12-04 22:05 796传送门:http://www.tektalk.org/2 ... -
有关装机与虚拟机
2011-12-15 14:44 0http://itbbs.pconline.com.cn/no ... -
hadoop 源代码学习
2011-12-15 14:44 0根据《hadoop 源代码分析 完整版》 (四) 主 ... -
hadoop io机制 序列化与反序列化
2011-12-15 11:27 0原文: http://blog.csdn.net/asx20 ... -
对象的序列化与反序列化
2011-12-15 11:02 0原文: http://www.blogjava.net/am ... -
source insight使用
2011-12-15 10:24 939原文: http://www.sourceinsight.c ... -
[转]Twitter首席工程师:如何“打败”CAP定理
2011-12-12 11:44 1886英文原文:How to beat the CAP theore ... -
自行编译hadoop eclipse-plugin for eclipse3.7
2011-12-12 11:33 2855主要是这两篇文章: http://varyall.iteye ... -
mapper,reducer的个数
2011-12-07 13:10 0《实战Hadoop》中第3章有一节讨论reducer的个数,从 ... -
hadoop eclipse plugin 之二
2011-12-03 11:47 1729前面已经有一篇记述了hadoop-plugin for ecl ... -
HDFS常用操作与API
2011-11-22 11:16 0《Hadoop》实践Chap2 相应条目下给出命令与贴图(另 ... -
转载:统计学读物推荐
2011-11-17 10:32 1098传送门:http://www.hadoopor.com/thr ... -
所有程序员都应该至少读上两遍的十篇论文
2010-12-31 00:11 666原文链接 http://blog.csdn.net/turi ... -
MapReduce:一个重大的倒退
2011-10-14 15:55 1128原文:http://apps.hi.baidu ... -
hbase单机终于跑起来了
2011-11-02 15:00 4169去看了日志文件 2011-11-01 23:44:06,99 ... -
hbase终于跑起来了 续
2011-11-02 15:09 880其实在前面还有一段,没有保存下当时的经历,现在将配置项保存一下 ...
相关推荐
全书分为10章,系统地介绍了HDFS存储系统,Hadoop的文件I/O系统,MapReduce2.0的框架结构和源码分析,MapReduce2.0的配置与测试,MapReduce2.0运行流程,MapReduce2.0高级程序设计以及相关特性等内容。《MapReduce...
全书分为10章,系统地介绍了HDFS存储系统,Hadoop的文件I/O系统,MapReduce 2.0的框架结构和源码分析,MapReduce 2.0的配置与测试,MapReduce 2.0运行流程,MapReduce 2.0高级程序设计以及相关特性等内容。...
##### 2.1 MapReduce 流程分析 - **输入文件处理**: - 读取输入文件内容,并将其解析为 key-value 对。 - 每一行数据都会被解析成 key-value 对,并传递给 map 函数处理。 - **map 任务处理**: - 实现自定义逻辑...
MapReduce是一种分布式数据处理模型,广泛应用于大数据的处理和分析领域。它的核心思想是将大任务分解成许多小任务,然后并行处理,最后再将结果进行汇总。在MapReduce模型中,主要包括两个阶段:Map阶段和Reduce...
##### 2.2 MapReduce流程分析 **2.2.1 Map过程** 1. **输入分片**:每个输入分片会被一个map任务处理,默认情况下以HDFS的一个块大小(通常是64MB)作为分片大小。可以通过调整HDFS配置文件中的相关参数来改变块的...
《基于MapReduce分析的招聘信息大数据可视化系统》 在当今数据驱动的时代,招聘信息的大数据...通过优化数据处理流程,提升数据分析的准确性和实时性,企业在人才竞争中能够更好地把握先机,实现人力资源的有效配置。
在本分析中,我们将深入MapReduce的源码,理解其内部机制,以及常用API的工作原理。 **Map阶段** Map阶段是MapReduce处理流程的第一步,它的主要任务是将输入数据分割成多个小块(split),然后对每个块执行用户...
### Hadoop.MapReduce 分析 #### 一、概述 Hadoop.MapReduce 是一种分布式计算模型,主要用于处理大规模数据集。其基本思想源自Google提出的MapReduce论文。本文将深入解析Hadoop.MapReduce的工作原理、核心组件...
在这个"MapReduce分析Youtube数据内含源码以及说明书可以自己运行复现.zip"压缩包中,包含了对YouTube数据进行分析的完整流程。这可能包括用户行为数据、视频信息、评论等。通过MapReduce,我们可以对这些数据进行...
在这个案例中,我们将深入探讨MapReduce的工作流程,以及如何通过代码来实现大数据的分析。 1. MapReduce的基本概念 MapReduce的核心思想是将大规模数据集分割成小块,然后在多台机器上并行处理这些小块。Map阶段...
map reduce的全部执行流程,源码分析视图
实验内容涵盖了从启动全分布模式的Hadoop集群到编写、运行和分析MapReduce应用程序的全过程。 首先,实验启动了Hadoop集群的所有守护进程,包括NameNode(主节点,负责元数据管理)、DataNode(存储数据的节点)、...
(3)对于大数据清洗和分析的基本操作流程 清洗不符合规范的数据以及不需要采用的特殊数据、通过字典简化数据格式(如实验第二步中用到的行为地址基础数据)、分析数据中对于生产或研究有意义的数据(如实验第五步...
* MapReduce 程序的执行流程 * MapReduce 程序的优化方法 MapReduce 项目实践 在实践中,我们可以使用 MapReduce 来解决各种大数据处理问题。以下是一些 MapReduce 项目实践: * WordCount 程序编写及代码分析 * ...
- **不适合迭代计算**:MapReduce不支持迭代式的算法,这限制了其在某些复杂数据分析场景中的应用。 - **内存管理**:虽然MapReduce能够处理大规模数据,但在处理特别大的数据集时可能会遇到内存不足的问题。 综上...