Hadoop代码分析（二）

Inputormat：describes the input-specification for a Map-Reduce job

?

它是一个用于处理指定输入的接口：

1.用于验证输入的有效性；

2.将指定输入切割成许多逻辑上的split供mapper函数处理；

3.提供一个RecodeReader让mapper一点点的处理split中的数据；

Inputformat其中有两个方法createRecodeReader（），getplits（），先看看getsplit（),将指定的输入切割成不同的split，有一个split基类叫做FIleInputFormat，所有的InputFormat都继承该类。大家都知道hadoop中一个block快是64M，代码中用blocksize标识它，getSplit的机制就是：先从minsize，maxsize，blocksize中计算Math.max(minSize, Math.min(maxSize, blockSize))，一般情况下splitsize都是blocksize（只要你的maxsize，minsize设置的别太多分就行了，呵呵～），然后循环的计算（byteremaining（length-splitsize的值）/splitsize）是否大于一个名为SPLIT_SLOP=1.1的值，如果大于，则生成一个FIleSplit：起始位置是上一个split的最后一个位置，长度是splitsize，第一次的起始位置当然是0（源码中用length-bytesRemaining标识），循环的最后将bytesRemaining将去该splitsize，然后继续循环；如果（byteremaining（length-splitsize的值）/splitsize）不大于1.1，并且bytesRemaining不为0，那么也将剩余的bytes生成一个split，其余的代码中也有，想必大家都会看得；

?

下面是源代码：

?

// generate splits    List<InputSplit> splits = new ArrayList<InputSplit>();    for (FileStatus file: listStatus(job)) {      Path path = file.getPath();      FileSystem fs = path.getFileSystem(job.getConfiguration());      long length = file.getLen();      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);      if ((length != 0) && isSplitable(job, path)) {         long blockSize = file.getBlockSize();        long splitSize = computeSplitSize(blockSize, minSize, maxSize);        long bytesRemaining = length;        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);          splits.add(new FileSplit(path, length-bytesRemaining, splitSize,                                    blkLocations[blkIndex].getHosts()));          bytesRemaining -= splitSize;        }                if (bytesRemaining != 0) {          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,                      blkLocations[blkLocations.length-1].getHosts()));        }      } else if (length != 0) {        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));      } else {         //Create empty hosts array for zero length files        splits.add(new FileSplit(path, 0, length, new String[0]));      }    }    LOG.debug("Total # of splits: " + splits.size());    return splits;  }

接下来是RecordRedaer,在源代码中，有一个名为LineRecordReader的类，用于从输入的split读入key/value对，

?