Hadoop编程模型之InputFormat

一、InputFormat接口简介

InputFormat接口主要用于描述输入数据的格式，它有两个方法：

InputSplit[] getSplits(JobConf job, int numSplits)

RecordReader<K, V> getRecordReader(InputSplit split,
JobConf job,
Reporter reporter)

getSplits方法完成数据切分功能，它会将输入数据切分成numSplits个InputSplit，注意InputSplit有两个特点：

1.只是逻辑上的分片，只记录分片的元数据信息，比如起始位置、长度、所在的节点列表

2.可序列化，以保证进程间通信。当作业被提交到JobTracker之前，Client会调用作业InputFormat中的getSplit函数，并将得到的InputSplit序列化到文件中，当作业提交到JobTracker端对作业初始化时，可直接读取该文件，解析出所有的InputSplit，并创建对应的Map Task

getRecorder方法返回一个RecordReader对象，该对象可将输入的InputSplit解析成若干个key/value对。MapReduce框架在Map Task执行过程中会不断调用RecordReader对象中的方法，迭代获取key/value对并交给map函数处理

二、InputFormat类图



所有基于文件的InputFormat实现的基类是FileInputFormat，这里详细介绍FileInputFormat

FileInputFormat是一个基类，它为各种InputFormat提供统一的getSplits函数，由派生类自己实现getRecordReader函数。FileInputFormat核心算法是文件切分算法和host选择算法

三、文件切分算法

文件切分算法主要用于确定InputSplit的个数以及每个InputSplit对应的数据段。FileInputFormat以文件为单位切分生成InputSplit，对于每个文件，有以下三个属性值确定其对应的InputSplit的个数

1.goalSize：根据用户期望的Inputplit数目计算出来的，即totalSize/numSplits。totalSize指文件总大小，numSplits指用户设定的Map Task个数，默认为1

2.minSize：InputSplit的最小值，由配置参数mapred.min.split.size确定，默认为1

3.blockSize：文件在HDFS中存储的block大小，不同文件可能不同，默认64MB

这三个参数共同决定InputSplit的最终大小

splitSize=max{minSize,min{goalSize,blockSize}}

确定splitSize后，FileInputFormat将文件依次切分为大小为splitSize的InputSplit，剩余不足的单独成为一个InputSplit

四、host选择算法

InputSplit切分方案确定后，就要确定InputSplit的元数据信息

<file,start,length,host>

，即InputSplit所在的文件、起始位置长度、和host列表。关键就在于host的选择方法，这个策略直接影响运行过程中的任务本地性。

在任务调度时，hadoop会优先让空闲资源处理本节点上的数据，如果节点上没有可处理的数据，则处理同一个机架上的数据，最差的情况是处理其他机架上的数据

虽然InputSplit对应的block可能位于多个节点上，但是考虑到任务调度的效率，通常不会把所有节点加到InputSplit的host列表中，它只选择前几个节点，以作为任务调度时判断任务是否具有本地性的主要凭证

FileInputFormat设计了一个简单有效的启发式算法：

1.按照rack包含的数据量对rack排序

2.在rack内部按照每个node包含的数据量对node排序

3.取前N（block的副本数）个node的host作为InputSplit的host列表。

这样，当任务调度器调度Task时，只要将Task调度给位于host列表的节点，即认为该Task满足本地性

结论：当使用基于FileInputFormat实现InputFormat时，为了提高Map Task的数据本地性，应尽量使InputSplit大小与block大小相同