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大小相同。