首页 > 代码库 > InputFormat&OutputFormat

InputFormat&OutputFormat

本文的主要目的是从源码级别讲解Hadoop中InputFormat和OutputFormat部分,首先简介InputFormat和OutputFormat,然后介绍两个重要的组件,RecordWriter和RecordReader,再以FileInputFormat和FileOutputFormat为例,介绍一组InputFormat和OutputFormat的实现细节,最后以SqoopInputFormat和SqoopOutputFormat为例,体会一下InputFormat和OutputFormat的灵活应用。

InputFormat和OutputFormat简介

InputFormat规定了MR框架,如何解析输入文件,比如说TextOutputFormat内部实现了一个LineRecordReader,从LineRecordReader#nextKeyValue方法可以看出,它的key是将要读的这一行数据的起始字节位置,value是这一行的内容。

OutputFormat规定了MR框架,最后输出的文件的内容格式,比如说SequenceFileOutputFormat,其getRecordWriter方法返回一个RecordWriter的内部类,具体的操作在SequenceFile类里write相关的方法中。SequenceFile的具体实现只从类来看比价复杂,其一条记录可以简单理解成如下的格式[record length][key length][key][value]。

RecordReader和RecordWriter简介

上边提到了RecordReader和RecordWriter,这里简单介绍一下,RecordWriter和RecordReader在MR框架每一个输入输出的地方用到,读写操作调用的都是RecordReader和RecordWriter提供的接口,比如说我们在Mapper#map和Reducer#reduce的结束通常都会写这样一行代码context.write(key,value),这行代码实际调用的就是RecordWriter#write方法,RecordReader是MR框架调用的。

一个InputFormat对应一个RecordReader,一个OutputFormat对应一个RecordWriter。这样在后边的DIY的时候,也知道应该写些什么了。

FileInputFormat和FileOutputFormat

这里以FileInputFormat,FileOutputFormat和其对应的子类TextInputFormat和TextOutputFormat为例,分析一套InputFormat和OutputFormat的具体实现。

1.      InputFormat

InputFormat抽象类中有两个方法:

InputFormat#getSplits(JobContext context)

对输入文件进行逻辑上的分片

InputFormat#createRecordReader(InputSplitsplit,TaskAttemptContext context)

返回一个RecordReader对象,该对象可将输入的InputSplit解析成若干个key/value对。

2.      FileInputFormat

所有需要处理文件的InputFormat的基类,从其实现可以看出,FileInputFormat主要是实现InputFormat#getSplit(JobContext context)方法,由于每个子类对应不同的输入格式,所以解析InputSplit的方法InputFormat#createRecordReader(InputSplit split,TaskAttemptContextcontext)由各个子类自己实现。

这里我们先分析FileInputFormat#getSplit(JobContext context)方法。

 

/**
  * Generate the list of files and make them into FileSplits.
  * @param job the job context
  * @throws IOException
  */
 public List<InputSplit> getSplits(JobContext job) throwsIOException {
   long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
   long maxSize = getMaxSplitSize(job);
 
// generatesplits
   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) {//文件不是空的
        FileSystem fs =path.getFileSystem(job.getConfiguration());
        BlockLocation[] blkLocations =fs.getFileBlockLocations(file, 0, length);//获取一个文件的所有的Block的带位置的信息
        if (isSplitable(job, path)) {
          long blockSize = file.getBlockSize();
          long splitSize =computeSplitSize(blockSize, minSize, maxSize);//计算一个FileSplit的大小,计算过程如下,Math.max(minSize,Math.min(maxSize, blockSize))
 
         long bytesRemaining =length;
                     //这个while循环就是根据上边的准备的信息,不停的读文件,产生FileSplit,一直到文件末尾。
          while (((double)bytesRemaining)/splitSize > SPLIT_SLOP) {
            int blkIndex =getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path,length-bytesRemaining, splitSize,
                                    blkLocations[blkIndex].getHosts()));
            bytesRemaining -= splitSize;
          }
                     //处理最后不足splitSize大小的数据
          if (bytesRemaining != 0) {
           int blkIndex =getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path,length-bytesRemaining, bytesRemaining,
                      blkLocations[blkIndex].getHosts()));
          }
        } else { // not splitable不能分割的话,文件只有一块blkLocations[0]就是这一块
          splits.add(makeSplit(path, 0, length,blkLocations[0].getHosts()));
        }
     } else {
        //Create empty hosts array for zerolength files
        splits.add(makeSplit(path, 0, length,new String[0]));
     }
   }
   // Save the number of input files for metrics/loadgen
   job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
   LOG.debug("Total # of splits: " + splits.size());
   return splits;
  }

3.      TextInputFormat

TextInputFormat直接使用了FileInputFormat的getSplits方法,自己实现了createRecordReader方法。

TextInputFormat#createRecordReader方法逻辑比较简单,最后返回一个LineRecordReader对象,下面主要看这个方法LineRecordReader#nextKeyValue。

public boolean nextKeyValue() throws IOException {
    if (key == null) {
      key = newLongWritable();
    }
    key.set(pos);
    if (value =http://www.mamicode.com/= null) {>

4.      OutputFormat

OutputFormat# getRecordWriter (TaskAttemptContext context)

返回一个RecordWriter对象

OutputFormat# checkOutputSpecs (JobContext context)

判断输出文件存不存在,写过MR程序的人应该很熟悉了,如果输出路径已经存在话,会抛出一个Output directory " + outDir + " already exists"错误,就是这个方法抛出的。

OutputFormat# getOutputCommitter (TaskAttemptContext context)

MR运行过程中,会产生很多中间文件,比如Mapper的输出,推测式执行Task时产生的文件等等,这个方法负责在任务执行完成后,处理这些中间文件。顺便说下,OutputCommitter对象里的方法都是回调方法,MR自动调用。

5.      FileOutputFormat

FileOutputFormat对上边提到的3个方法中的后两个提供了通用的实现,OutputFormat# getRecordWriter (TaskAttemptContext context)方法需要子类自己实现。

public void checkOutputSpecs(JobContext job
                              ) throws FileAlreadyExistsException, IOException{
    // Ensure that the outputdirectory is set and not already there
    Path outDir =getOutputPath(job);
    if (outDir == null) {
      throw newInvalidJobConfException("Output directory not set.");
    }
 
    // get delegation tokenfor outDir's file system
   TokenCache.obtainTokensForNamenodes(job.getCredentials(),
        new Path[] { outDir },job.getConfiguration());
 
    if(outDir.getFileSystem(job.getConfiguration()).exists(outDir)) {
      throw newFileAlreadyExistsException("Output directory " + outDir +
                                          " already exists");
    }
  }
 
public synchronized
     OutputCommittergetOutputCommitter(TaskAttemptContext context
                                        )throws IOException {
    if (committer == null) {
      Path output =getOutputPath(context);
      committer = newFileOutputCommitter(output, context);
    }
    return committer;
  }

6.      TextOutputFormat

TextOutputFormat直接使用了FileInputFormat的getSplits方法,自己实现了createRecordReader方法。

TextOutputFormat#createRecordWriter方法逻辑比较简单,最后返回一个LineRecordWriter对象,下面主要看这个方法LineRecordReader#write(K key,V Value)。

public synchronized void write(K key, V value)
      throws IOException {
 
      boolean nullKey = key ==null || key instanceof NullWritable;
      boolean nullValue =http://www.mamicode.com/value == null || value instanceof NullWritable;>

SqoopInputFormat和SqoopOutputFormat

下面的分析基于Sqoop1.99.3

在Sqoop中,InputFormat和OutputFormat的子类有3个,分别是:

public class SqoopInputFormat extendsInputFormat<SqoopSplit, NullWritable>
public class SqoopFileOutputFormat extendsFileOutputFormat<Data, NullWritable>
public class SqoopNullOutputFormat extendsOutputFormat<Data, NullWritable>

下面一个个分析:

1.      SqoopInputFormat

从泛型传入的类型可以看出,SqoopInputFormat的key是SqoopSplit,value是NullWritable。直接把整个InputSplit作为key,传到Mapper中,SqoopInputFormat不对InputSplit做任何的解析操作。

2.      SqoopFileOutputFormat

从泛型传入的类型来看,跟SqoopInputFormat类似,只是用KEY部分保存信息。SqoopFileOutputFormat只重写了父类FileOutputFormat的getRecordWriter方法和getOutputCommitter方法,checkOutputSpec方法使用的父类的。

getRecordWriter方法调用SqoopOutputFormatLoadExecutor#getRecordWriter,这个方法在返回一个SqoopRecordWriter的同时,开启一个消费者线程,SqoopRecordWriter是生产者线程。[后面接着分析Sqoop源码时细说]

3.      SqoopNullOutputFormat

SqoopNullOutputFormat将OutputFormat的3个方法都重写了。SqoopNullOutputFormat#getRecordWriter方法同样是调用SqoopOutputFormatLoadExecutor#getRecordWriter。

InputFormat&OutputFormat