关于Hadoop的InputFormat类

 

org.apache.hadoop.mapreduce.InputFormat 是一个抽象类，

关于这个抽象类的功能描述如下：

1、首先为Job验证输入；

2、将输入的文件分成逻辑上的splits，每个split会被应用到一个单独的mapper上；

3、提供RecorderReader的实现，用来从逻辑split中一点一点的收集数据到mapper中。

 

 

这个抽象类中，有两个抽象方法需要实现：

1.   public abstract List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException;

从方法名和它的返回类型就可以猜到，这个接口主要责任就是将一大堆的输入文件分成一系列的splits（每个split是用一个InputSplit对象来表示）。然后每个InputSplit被传递给不同的mapper进行处理。但是要注意，分片仅仅是逻辑上的分片，并不是真地将文件分成多块了。一个分片可以用（输入文件路径，开始位置，偏移量）元组来表示。

2.   public abstract RecordReader<K,V> createRecordReader(InputSplit split,  TaskAttemptContext context ) throws IOException,  InterruptedException;

这个接口的责任是返回一个读取器，来读取这种InputFormat的分片文件，至于怎么读取，就是一种读取策略了，Hadoop框架自身实现了一些，我们当然也可以实现自己的策略满足自己的需求。

 

       在Hadoop中最常用的就是文件作为 job 的输入，这个是由抽象类 FileInputFormat 和 其子类来实现的，它将输入的文件按照大小进行分片。文件系统的块大小被看作是分片的一个上界。下界可以通过设定mapred.min.split.size来指定。基于文件大小的这种逻辑分片方法有的时候是低效的，因为这个时候我们必须去关注边界上的记录是否完整并做出特殊处理，只有这样处理之后，确保没有中间截断的记录，这样才能够传递给mapper来进行进一步的处理。

 

       其中的FileInputFormat继承自InputFormat，但是只是实现了getSplits方法，另一个获取读取器的方法没有实现，这样做是有道理的，因为很多不同格式的文件需要使用不同的读取器来提取数据，比如lzo压缩后的文件的读取器，要先解压后才能读取。

 

源代码中形成splits列表的逻辑大概是这样的：

首先会从 job 对象中所有的输入文件的列表提取出来

List<FileStatus>files = listStatus(job);

然后就要对每个文件进行逻辑分片了，

 

分片的逻辑大概是这样的：

        首先计算这个文件的长度（按照字节），然后将这个文件的块信息拿出来。如果这个文件可以被分片并且长度不是0，那么就开始进行逻辑分片。每个分片的大小通过函数computeSplitSize来计算。然后如果文件的剩余长度是分块的1.1倍以上的话，就创建一个新分片：

splits.add(new FileSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));

进而，将剩余长度减去已经被分配掉的splitSize，这样循环直到不满足条件。等循环完成之后，如果还有剩余的部分，那么剩下就可以再做一个分片，加入到列表中。

但是，如果我们一开始输入的文件的大小是不可分割的话，那么我们就把整个文件作为一个分片，形成一个实例：

splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));

如果这个文件是可分割的，但是长度是0，也做一个默认的分片：

splits.add(new FileSplit(path, 0, length, new String[0]));

这样，文件的分片列表就产生了，然后读取器就可以从这些分片中按照相应的读取逻辑来读取数据，并交给mapper进行处理了。