标签：hadoop-2.4.1 源代码 inputformat recordreader inputsplit
向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等将只做简要描述。
FileInputFormat的主要子类有：TextInputFormat、SequenceFileInputFormat、NLineInputFormat、KeyValueTextInputFormat、FixedLengthInputFormat和CombineFileInputFormat，下面简要概述其用途和特点。
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是如何实现读取数据的可以阅读其源代码。
Hadoop-2.4.1学习之InputFormat及源代码分析
标签：hadoop-2.4.1 源代码 inputformat recordreader inputsplit
原文地址：http://blog.csdn.net/skywalker_only/article/details/42678659
