首页 > 代码库 > Hadoop-2.4.1学习之InputFormat及源代码分析

Hadoop-2.4.1学习之InputFormat及源代码分析

      向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。

      在综合概述了InputFormat及其子类的用途后,下面结合源代码具体分析FileInputFormat中如何分割逻辑InputSplit及如何使用RecordReader读取数据的。下面是getSplits的源代码:

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及源代码分析