Hadoop-2.4.1学习之InputFormat及源代码分析，hadoop2.4.1
Hadoop-2.4.1学习之InputFormat及源代码分析，hadoop2.4.1
向Hadoop集群提交作业时，需要指定作业输入的格式（未指定时默认的输入格式为TextInputFormat）。在Hadoop中使用InputFormat类或InputFormat接口描述MapReduce作业输入的规范或者格式，之所以说InputFormat类或InputFormat接口是因为在旧的API（hadoop-0.x）中InputFormat被定义为接口，而在新的API（hadoop-1.x及hadoop-2.x）中，InputFormat是做为抽象类存在的，在本篇文章中主要讲述InputFormat抽象类及其子类。InputFormat主要用于验证作业的输入是否符合规范，将输入文件分割为逻辑InputSplit，每个InputSplit被分配给一个Mapper任务，提供RecordReader的实现，该实现负责从InputSplit中收集记录交由Mapper任务处理。不同的InputFormat子类提供了不同的InputSplit和RecordReader，比如用于文件的FileSplit和用于数据库的DBInputSplit，用于文本文件的LineRecordReader和Sequence文件的SequenceFileRecordReader等。下图为InputFormat及其子类关系图：
InputFormat抽象类只提供了两个抽象方法，分别用于获取InputSplit和RecordReader。其中的InputSplit只是对输入文件的逻辑分割，而不是物理上将输入文件分割为块，或者说InputSplit只是指定了输入文件的某个范围输入到特定的Mapper中。在实际的应用中，大多数情况都是使用FileInputFormat的子类做为输入，故此在本篇文章中将重点学习FileInputFormat及其子类，对于其它类比如DBInputFormat等将只做简要描述。
- DBInputFormat：从SQL表中读取数据的InputFormat。DBInputFormat使用包含记录号的LongWritable做为键，DBWritable做为值。该类的子类为DataDrivenDBInputFormat，该类与父类使用不同的机制划分InputSplit。
- ComposableInputFormat：：抽象类，该类的子类需要提供ComposableRecordReader而不再是RecordReader。相对于RecordReader，ComposableRecordReader提供了额外的方法。
- CompositeInputFormat：在具有相同的排序和分区的一组数据源上执行join操作的InputFormat。
- FileInputFormat：抽象类，是所有基于文件输入的InputFormat的基础类。该类提供了getSplits(JobContext)的通用实现，其子类也可以通过覆盖isSplitable(JobContext, Path)确定输入文件是否可以分割还是整个输入文件都将被特定Mapper任务处理。
FileInputFormat的主要子类有：TextInputFormat、SequenceFileInputFormat、NLineInputFormat、KeyValueTextInputFormat、FixedLengthInputFormat和CombineFileInputFormat，下面简要概述其用途和特点。
- TextInputFormat：用于纯文本文件的InputFormat，也是默认的InputFormat。输入文件被分解为行，回车或者换行做为行结束的标记，键为行在文件中的位置，值为行内容。该InputFormat使用LineRecordReader读取InputSplit的内容。
- NLineInputFormat：将输入中的N行做为一个InputSplit，其中N可以由参数mapreduce.input.lineinputformat.linespermap指定，默认为1。RecordReader也是用LineRecordReader。
- KeyValueTextInputFormat：用于纯文本文件的InputFormat，每行使用分隔字节划分为键和值，该分隔符由参数mapreduce.input.keyvaluelinerecordreader.key.value.separator指定，默认使用\t。如果该分隔符不存在则整行将做为键，值为空。RecordReader为KeyValueLineRecordReader。
- FixedLengthInputFormat：用于读取文件中的记录为固定长度的InputFormat，文件的内容不是必须为文本，也可以是二进制数据。用户必须通过参数fixedlengthinputformat.record.length设置记录的长度，默认值为0，若参数的值小于等于0，将会抛出IOException。RecordReader为FixedLengthRecordReader。
- CombineFileInputFormat：抽象类，该类的getSplits(JobContext)返回的不是List<FileSplit>而是List<CombineFileSplit>。CombineFileSplit根据输入路径中的文件构造，该对象不能够有不同池中的文件，每个CombineFileSplit可能包含不同文件的块。该类有两个子类用于sequence文件和纯文本文件，分别为CombineSequenceFileInputFormat和CombineTextInputFormat，其RecordReader分别为SequenceFileRecordReaderWrapper和 TextRecordReaderWrapper。
- SequenceFileInputFormat：用于sequence文件的InputFormat，获取InputSplit的方法继承自FileInputFormat，并未重写，其RecordReader为SequenceFileRecordReader。该类有三个子类：SequenceFileAsBinaryInputFormat、SequenceFileAsTextInputFormat和 SequenceFileInputFilter，分别用于从sequence文件的二进制格式中读取键值、将sequence文件中的键值转换为字符串形式、从sequence文件中抽样然后交由MapReduce作业处理，抽样由过滤器类确定。三者的RecordReader分别为：SequenceFileAsBinaryRecordReader、SequenceFileAsTextRecordReader和FilterRecordReader。
public List<InputSplit> getSplits(JobContext job) throws IOException {
//以纳秒为单位测试执行的时间
Stopwatch sw = new Stopwatch().start();
//取特定格式的最小分片大小和mapreduce.input.fileinputformat.split.minsize设置的值二者中的较大者，默认为1
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
//参数mapreduce.input.fileinputformat.split.maxsize的值，默认为Long.MAX_VAULE
long maxSize = getMaxSplitSize(job);
// generate splits
List<InputSplit> splits = new ArrayList<InputSplit>();
//获取作业的输入文件
List<FileStatus> files = listStatus(job);
for (FileStatus file: files) {
Path path = file.getPath();
long length = file.getLen();
if (length != 0) {
BlockLocation[] blkLocations;
//获取file表示的文件所属块的位置
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
FileSystem fs = path.getFileSystem(job.getConfiguration());
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
if (isSplitable(job, path)) {
//获取文件块大小
long blockSize = file.getBlockSize();
//计算InputSplit大小，通常返回的值为dfs.blocksize的值
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
long bytesRemaining = length;
//若剩余值大于1.1*splitSize，则继续对文件划分，若小于等于该值，则做为一个InputSplit
//也就是说每个InputSplit的最大值为1.1*splitSize，最小文件至少大于0.1*splitSize
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
//计算文件分块的索引，此处只计算InputSplit的起始位置是否位于某个块中
//而不管InputSplit的大小是否会超出该块的范围（InputSplit是逻辑概念）
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
//起始位置为0，splitSize，2splitSize……，length为splitSize
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts()));
}
} else { // not splitable
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files for metrics/loadgen
//设置mapreduce.input.fileinputformat.numinputfiles的值为输入文件的数量
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.elapsedMillis());
}
return splits;
}
在上面的代码中，如果splitSize等于blocksize，则InputSplit的起始位置与相应块在文件中offset一致，如果splitSize小于blocksize，即通过参数mapreduce.input.fileinputformat.split.maxsize控制每个InputSplit的大小，那么InputSplit的起始位置就会位于文件块的内部。上述分割InputSplit的逻辑完全是针对大小进行的，那么是否存在将一行记录划分到两个InputSplit中的可能性？答案是肯定的。但上述代码并未对这种情况进行处理，这也意味着在Map阶段存在数据不完整的可能，Hadoop当然不会允许这种情况的发生，而RecordReader就负责处理这种情况，下面以LineRecordReader为例，看看Hadoop是如何处理记录跨InputSplit的。LineRecordReader中的initialize方法在初始化的时候调用一次，此方法将定位InputSplit中第一个换行符的位置，并将实际读取数据的位置定位到第一个换行符之后的位置，如果是第一个InputSplit则不用如此处理。而被过滤的内容则由读取该InputSplit之前的Split的LineRecordReader负责读取，确定第一个换行符位置的代码片段为（此处假设输入为未压缩文件）：
//定位到该FileSplit在输入文件中的起始位置
fileIn.seek(start);
in = new SplitLineReader(fileIn, job, this.recordDelimiterBytes);
filePosition = fileIn;
// If this is not the first split, we always throw away first record
// because we always (except the last split) read one extra line in
// next() method.
//如果不是第一个split（第一个split的start为输入文件的开头，即0），则忽略该split中第一个换行符及之前的数据
//这些数据可能是一整行数据，可能是一行数据的部分内容，也可能仅是换行符本身
if (start != 0) {
//readLine从fileIn读取到换行符（CR，LF以及CRLF）的长度
//然后将start定位到下一行的开始
start += in.readLine(new Text(), 0, maxBytesToConsume(start));
}
this.pos = start;
读取InputSplit开头内容的代码位于nextKeyValue中，具体代码为：
int newSize = 0;
// We always read one extra line, which lies outside the upper
// split limit i.e. (end - 1)
//此处将读取下一个InputSplit的数据直到换行符，所以initialize需要将下一个InputSplit中第一个换行符之前的内容去掉
while (getFilePosition() <= end || in.needAdditionalRecordAfterSplit()) {
newSize = in.readLine(value, maxLineLength,
Math.max(maxBytesToConsume(pos), maxLineLength));
pos += newSize;
if (newSize < maxLineLength) {
break;
}
// line too long. try again
LOG.info("Skipped line of size " + newSize + " at pos " +
(pos - newSize));
}
readLine的功能是读取一行的数据到Text中，因为在分割FileSplit时是基于size的，如果一行被分割到两个split中，比如s1和s2中，在读取s1中的最后一行数据时，会一直读取到s2中的第一个换行符，这是在nextKeyValue方法中实现的，而在处理s2时，则要将已经读取的数据跳过以避免重复读取，这是在initialize中实现的。其它类型的RecordReader是如何实现读取数据的可以阅读其源代码。
评论暂时关闭