MapReduce编程模型之InputFormat接口分析 博客分类: hadoop MapReduce编程模型InputFormat
最近小组任务是在MapReduce编程模型,分析其接口体系结构,即InputFormat、Mapper、Partitioner、Reducer和OutputFormat五个接口的分析,我的任务是分析InputFormat接口,当时觉得不就是几个接口吗,还用得着五个人来分工,回家一晚上就可以全部搞定,好在当时没有冲动,现在花了一周(其实只有假期里)总算勉强玩转一个,想想还有点小激动。
想深入学习MapReduce的小伙伴们不妨看下hadoop技术内幕,有两册,楼主看的是MapReduce的架构设计和实现原理,还有Commen+HDFS,后者对Commen和HDFS的源码进行了分析,感兴趣的话read下也是极好的。再者就是谷歌三大论文中有关于MapReduce的论文,百度上都有PDF文档可以直接下载,嫌麻烦的同学可以看下http://blog.csdn.net/active1001/article/details/1675920,其实内容都差不多啦。
MapReduce应用广泛的原因之一在于它的易用性。它提供了一个因高度抽象化而变得异常简单的编程模型。MapReduce是在总结大量应用的共同特点的基础上抽象出来的分布式计算框架,特点:任务可以分解成相互独立子问题。
上面灰色部分是整个编程模型,分为两层:所谓工具层就是增加点兼容性(楼主个人理解,不喜勿喷)而已,而下面的接口层就是费神的地方了。
先来点官方的,InputFormat主要用于描述数据的格式,它提供一下两个功能:
数据切分:按照某个策略将输入数据切分成若干个split,以便确定Map Task个数以及对应的split。
为Mapper提供输入数据:给定某个split,能将其解析成一个个key/value对。
乍一看,挺简单的,不就是把数据打碎,变成k&v对提交给Mapper么,好吧,学了等于没学,这些貌似没什么用。
楼主这里从数据划分、splitI调度和数据读取这三方面来分析InputFormat。
具体还是先从经典的MapReduce工作流程入手吧。
1、运行一个MapReduce程序;
2、运行时将生成一个Job,JobClient向JobTracker申请一个JobID以标识这个Job;
3、JobClient将Job所需要的资源提交到HDFS中一个以JobID命名的目录中。这些资源包括JAR包、配置文件、InputSplit等;
4、JobClient向JobTracker提交这个Job;
5、JobTracker接收并初始化这个Job;
6、JobTracker从HDFS获取这个Job的Split等信息;
7、JobTracker向TaskTracker分配任务;
8、TaskTracker从HDFS获取这个Job的相关资源;
9、TaskTracker开启一个新的JVM;
10、TaskTracker用新的JVM来执行Map或Reduce;
数据划分
注意看第三步,这里提交的资源中包含了InputSplit(上文中已标记)就是数据划分的结果。数据划分是在JobClient上完成的,它适用InputFormat将输入数据做一次划分,形成若干split。
InputFormat是一个接口,它包含两种方法:
1) InputSplit[] getsSplits(JobConf job,int numSplits) throws IOException;
2) RecordReader<K,V> getRecordReader(InpputSplit split,JobConf job,Reporter reporter) throws IOException;
这里getSplits函数就是划分函数。job参数是任务的配置集合,从中可以取到用户在启动MapReduce时指定的输入文件路径。而numSplits参数是一个Split数目。
返回接口数组InputSplit,描述所有的Split信息,一对一。它也有两个函数:
1) long getLength() throws IOException;
2) String[] getLocation() throws IoException;
即描述Split长度与Split的位置(在HDFS上存放的机器),属于逻辑分片。
序列化问题:对象序列化主要用于进程间通信和永久存储。InputSplit序列化主要是为了前者。
getRecordReader方法返回一个RecordReader对象,将输入的InputSplit解析成若干key/value对。
顺便插一句,有时候会遇到输入文件是不可划分的,比如一个tar.gz,划分会是其无法解压,可以考虑重载FileInoutFormat的isSplitable()函数来告知文件是否可分,或者干脆从头实现一个InputFormat。
Split调度
图片第6步JobTracker会从HDFS获取Job的Split信息,这将生成一系列待处理的Map和Reduce任务。 JobTracker 并不会主动的为每一个TaskTracker划分一个任务子集,而是直接把所有任务都放在跟Job对
应的待处理任务列表中。
在分配Map任务时,Split的Location信息就要发挥作用了。JobTracker会根据TaskTracker的地址来选择一个Location与之最接近的Split所对应的Map任务(注意一个Split可以有多个Location)。这样一来,输入文件中Block的Location信息经过一系列的整合(by InputFormat)和传递,最终就影响到了Map任务的分配。其结果是Map任务倾向于处理存放在本地的数据,以保证效率。
数据读取
图片第10步,TaskTracker启动一个JVM来执行Map程序。在Map执行的时候,会使用InputFormat.getRecordReader()所返回的RecordReader对象来读取Split中的每一条记录。
PS:实际上,RecordReader并不关心Split的Location,只管Open它的Path。前面说RecordReader是由具体的InputFormat创建并返回的,它跟对应的InputFormat所使用的InputSplit必定是配对的。比如,对应于 FileSplit,RecordReader要读取FileSplit.file文件中的相应区间、对于CombineFileSplit,RecordReader要读 取CombineFileSplit.paths中的每个文件的相应区间。
最后,介绍大家系统自带的各种InputFormat实现。
如上图,所有基于文件的InputFormat实现的基类是FileInputFormat,并派生出针对文本格式的TextInputFormat、KeyValueTextInputFormat和NLineInputFormat以及针对二进制文件格式的SequenceFileInputFormat等。
注意:当使用基于FileInputFormat实现InputFormat时,为了提高Map Task 的数据本地性,应尽量是InputSplit与block大小相同。