Hadoop源代码分析（包mapreduce.lib.input）

接下来咱们按照MapReduce过程当中数据流动的顺序，来分解org.apache.hadoop.mapreduce.lib.*的相关内容，并介绍对应的基类的功能。首先是input部分，它实现了MapReduce的数据输入部分。类图以下：

 

类图的右上角是InputFormat，它描述了一个MapReduceJob的输入，经过InputFormat，Hadoop能够：

l          检查MapReduce输入数据的正确性；

l          将输入数据切分为逻辑块InputSplit，这些块会分配给Mapper；

l          提供一个RecordReader实现，Mapper用该实现从InputSplit中读取输入的<K,V>对。

在org.apache.hadoop.mapreduce.lib.input中，Hadoop为全部基于文件的InputFormat提供了一个虚基类FileInputFormat。下面几个参数能够用于配置FileInputFormat：

l          mapred.input.pathFilter.class：输入文件过滤器，经过过滤器的文件才会加入InputFormat；

l          mapred.min.split.size：最小的划分大小；

l           mapred.max.split.size：最大的划分大小；

l          mapred.input.dir：输入路径，用逗号作分割。

类中比较重要的方法有：

  protectedList<FileStatus> listStatus(Configuration job)

递归获取输入数据目录中的全部文件（包括文件信息），输入的job是系统运行的配置Configuration，包含了上面咱们提到的参数。

  publicList<InputSplit> getSplits(JobContext context)

将输入划分为InputSplit，包含两个循环，第一个循环处理全部的文件，对于每个文件，根据输入的划分最大/最小值，循环获得文件上的划分。注意，划分不会跨越文件。
FileInputFormat没有实现InputFormat的createRecordReader方法。
FileInputFormat有两个子类，SequenceFileInputFormat是Hadoop定义的一种二进制形式存放的键/值文件（参考http://hadoop.apache.org/core/do ... o/SequenceFile.html），它有本身定义的文件布局。因为它有特殊的扩展名，因此SequenceFileInputFormat重载了listStatus，同时，它实现了createRecordReader，返回一个SequenceFileRecordReader对象。TextInputFormat处理的是文本文件，createRecordReader返回的是LineRecordReader的实例。这两个类都没有重载FileInputFormat的getSplits方法，那么，在他们对于的RecordReader中，必须考虑FileInputFormat对输入的划分方式。
FileInputFormat的getSplits，返回的是FileSplit。这是一个很简单的类，包含的属性（文件名，起始偏移量，划分的长度和可能的目标机器）已经足以说明这个类的功能。
RecordReader用于在划分中读取<Key,Value>对。RecordReader有五个虚方法，分别是：

l          initialize：初始化，输入参数包括该Reader工做的数据划分InputSplit和Job的上下文context；

l          nextKey：获得输入的下一个Key，若是数据划分已经没有新的记录，返回空；

l          nextValue：获得Key对应的Value，必须在调用nextKey后调用；

l          getProgress：获得如今的进度；

l          close，来自java.io的Closeable接口，用于清理RecordReader。

咱们以LineRecordReader为例，来分析RecordReader的构成。前面咱们已经分析过FileInputFormat对文件的划分了，划分完的Split包括了文件名，起始偏移量，划分的长度。因为文件是文本文件，LineRecordReader的初始化方法initialize会建立一个基于行的读取对象LineReader（定义在org.apache.hadoop.util中，咱们就不分析啦），而后跳过输入的最开始的部分（只在Split的起始偏移量不为0的状况下进行，这时最开始的部分多是上一个Split的最后一行的一部分）。nextKey的处理很简单，它使用当前的偏移量做为Key，nextValue固然就是偏移量开始的那一行了（若是行很长，可能出现截断）。进度getProgress和close都很简单。

更多精彩内容请关注：http://bbs.superwu.cn

关注超人学院微信二维码：