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对，