十六、FileInputFormat介绍，切片源码分析

一、InputFormat介绍

InputFormat，从单词意思解读分为输入、格式，也就是数据来源与加载数据的方式是决定MR编程的map阶段的任务并行度。

数据来源划分：其实也就是他的子类，由于我目前只使用了如下三种方式，其实还有很多子类。

HLogInputFormat：从hbase加载数据编写mr程序计算
FileInputFormat：主要从hdfs或本地加载数据
自定义实现：可以编写从mysql或oracle中加载数据

InputFormat它是一个抽象类，定义了获取切片与切片划分的抽象函数，具体工作有它的子类来完成。

public abstract class InputFormat<K, V> {public InputFormat() {}//切片：确定每个片的大小public abstract List<InputSplit> getSplits(JobContext var1) throws IOException, InterruptedException;//创建数据片的数据加载：读取当前切片的数据public abstract RecordReader<K, V> createRecordReader(InputSplit var1, TaskAttemptContext var2) throws IOException, InterruptedException;
}

什么是切片？

大家看到了上面提到的切片，相信会疑问什么是切片，它有什么意义？

大家都知道MR是分布是计算，那么它肯定是可以并行多个MapTask，但是MapTask的并行度到底是多少这就不清楚了，如下问题：

思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。
那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask
并行任务是否越多越好呢？哪些因素影响了MapTask并行度？

MapTask并行度决定机制：

数据块：Block是HDFS物理上把数据分成一块一块。
数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。

具体切片工作机制请看下图：

逻辑分片： 假设某个文件有200M，默认按照128M切片，那么就开启两个maptask，一个加载0-128M的数据，一个加载129M-200M的数据。并不是将数据切分为两个文件，这就叫逻辑分片，也就是逻辑切片。

二、FileInputFormat 介绍

相信编写mr程序的都不陌生，它是加载hdfs/本地数据的输入格式，其实它也是一个抽象类，mr运行时具体工作由它子类完成。但是它相比InputFormat完善了一些工作。InputFormat只是定义了数据切片，FileInputFormat完善了具体切片方式，下面就来介绍FileInputFormat切片机制。

1、FileInputFormat切片机制

（1）简单地按照文件的内容长度进行切片
（2）切片大小，默认等于Block大小
（3）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

具体切片过程看上图

2、切片大小设置

1）源码中默认计算切片公式

Math.max(minSize, Math.min(maxSize, blockSize));
mapreduce.input.fileinputformat.split.minsize=1 默认值为1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue因此，默认情况下，切片大小=blocksize。
完整代码：protected long computeSplitSize(long blockSize, long minSize, long maxSize) {return Math.max(minSize, Math.min(maxSize, blockSize));}

（2）切片大小设置

maxsize（切片最大值）：参数如果调得比blockSize小，则会让切片变小，而且就等于配置的这个参数的值。minsize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blockSize还大。

（3）获取切片信息API

// 获取切片的文件名称
String name = inputSplit.getPath().getName();
// 根据文件类型获取切片信息
FileSplit inputSplit = (FileSplit) context.getInputSplit();

3、FileInputFormat实现类

4、源码分析注释

//分片机制
public List<InputSplit> getSplits(JobContext job) throws IOException {Stopwatch sw = (new Stopwatch()).start();long minSize = Math.max(this.getFormatMinSplitSize(), getMinSplitSize(job));long maxSize = getMaxSplitSize(job);//分的每个片集合List<InputSplit> splits = new ArrayList();//获取当前路径下的所有文件List<FileStatus> files = this.listStatus(job);//迭代每个文件Iterator i$ = files.iterator();while(true) {while(true) {while(i$.hasNext()) {FileStatus file = (FileStatus)i$.next();Path path = file.getPath();long length = file.getLen();//判断文件是否有为空if (length != 0L) {//文件不为空BlockLocation[] blkLocations;if (file instanceof LocatedFileStatus) {blkLocations = ((LocatedFileStatus)file).getBlockLocations();} else {FileSystem fs = path.getFileSystem(job.getConfiguration());blkLocations = fs.getFileBlockLocations(file, 0L, length);}//判断文件是否分片if (this.isSplitable(job, path)) {long blockSize = file.getBlockSize();long splitSize = this.computeSplitSize(blockSize, minSize, maxSize);long bytesRemaining;int blkIndex;//遍历为文件分片，按照字节偏移量for(bytesRemaining = length; (double)bytesRemaining / (double)splitSize > 1.1D; bytesRemaining -= splitSize) {blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);splits.add(this.makeSplit(path, length - bytesRemaining, splitSize, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));}//如果文件大小不是片的倍数，则余下一块比块小的部分单独作为一片if (bytesRemaining != 0L) {blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));}} else {//如果文件过小则不用分片splits.add(this.makeSplit(path, 0L, length, blkLocations[0].getHosts(), blkLocations[0].getCachedHosts()));}} else {//splits.add(this.makeSplit(path, 0L, length, new String[0]));}}job.getConfiguration().setLong("mapreduce.input.fileinputformat.numinputfiles", (long)files.size());sw.stop();if (LOG.isDebugEnabled()) {LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.elapsedMillis());}return splits;}}}

1、获取路径下所有文件
2、迭代处理每个文件
3、判断文件大小是否为空，为空则创建一个片信息
4、文件不为空则判断文件是否需要分片
5、文件不为空且小于128M则直接作为一个片
6、文件大于128M，则按照字节偏移量为文件分片
7、文件分片完成后最后一部分小于128M的数据单独作为一片