码迷，mamicode.com
首页 > 其他好文 > 详细

Hadoop InputFormat源码分析

时间：2016-01-31 21:16:50      阅读：338      评论：0      收藏：0      [点我收藏+]

标签：

平时我们写MapReduce程序的时候，在设置输入格式的时候，总会调用形如job.setInputFormatClass(KeyValueTextInputFormat.class)来保证输入文件按照我们想要的格式被读取。所有的输入格式都继承于InputFormat，这是一个抽象类，其子类有专门用于读取普通文件的FileInputFormat，用来读取数据库的DBInputFormat等等。

技术分享

不同的InputFormat都会按自己的实现来读取输入数据并产生输入分片，一个输入分片会被单独的MapTask作为数据源，下面我们先看看这些输入分片(InputSplit)是什么样的。

 

InPutSplit：

我们知道Mapper的输入是一个一个的输入分片，称为InputSplit。InputSplit是一个抽象类，它在逻辑上包含了提供给处理这个InputSplit的Mapper的所有K-V对。

 

[java] view plain copy
 
  1. public abstract class InputSplit {  
  2.   
  3.   public abstract long getLength() throws IOException, InterruptedException;  
  4.   
  5.   
  6.   public abstract   
  7.     String[] getLocations() throws IOException, InterruptedException;  
  8. }  

getLength()用来获取InputSplit的大小，以支持对InputSplit进行排序，而getLocations()则用来获取存储分片的位置列表。

 

 

我们来看一个简单的InputSplit子类：FileSplit，源码如下：

 

[java] view plain copy
 
  1. public class FileSplit extends InputSplit implements Writable {  
  2.   private Path file;  
  3.   private long start;  
  4.   private long length;  
  5.   private String[] hosts;  
  6.   
  7.   FileSplit() {}  
  8.   
  9.   /** Constructs a split with host information 
  10.    * 
  11.    * @param file the file name 
  12.    * @param start the position of the first byte in the file to process 
  13.    * @param length the number of bytes in the file to process 
  14.    * @param hosts the list of hosts containing the block, possibly null 
  15.    */  
  16.   public FileSplit(Path file, long start, long length, String[] hosts) {  
  17.     this.file = file;  
  18.     this.start = start;  
  19.     this.length = length;  
  20.     this.hosts = hosts;  
  21.   }  
  22.    
  23.   /** The file containing this split‘s data. */  
  24.   public Path getPath() { return file; }  
  25.     
  26.   /** The position of the first byte in the file to process. */  
  27.   public long getStart() { return start; }  
  28.     
  29.   /** The number of bytes in the file to process. */  
  30.   @Override  
  31.   public long getLength() { return length; }  
  32.   
  33.   @Override  
  34.   public String toString() { return file + ":" + start + "+" + length; }  
  35.   
  36.   ////////////////////////////////////////////  
  37.   // 序列化和反序列化  
  38.   ////////////////////////////////////////////  
  39.   
  40.   @Override  
  41.   public void write(DataOutput out) throws IOException {  
  42.     Text.writeString(out, file.toString());  
  43.     out.writeLong(start);  
  44.     out.writeLong(length);  
  45.   }  
  46.   
  47.   @Override  
  48.   public void readFields(DataInput in) throws IOException {  
  49.     file = new Path(Text.readString(in));  
  50.     start = in.readLong();  
  51.     length = in.readLong();  
  52.     hosts = null;  
  53.   }  
  54.   
  55.   @Override  
  56.   public String[] getLocations() throws IOException {  
  57.     if (this.hosts == null) {  
  58.       return new String[]{};  
  59.     } else {  
  60.       return this.hosts;  
  61.     }  
  62.   }  
  63. }  

从上面的源码我们可以看到，一个FileSplit是由文件路径，分片开始位置，分片大小和存储分片数据的hosts列表组成，由这些信息我们就可以从输入文件中切分出提供给单个Mapper的输入数据。这些属性会在Constructor设置，我们在后面会看到这会在InputFormat的getSplits()构造这些分片。

 

 

我们再来看看CombinerFileSplit的源码：

 

[java] view plain copy
 
  1. @InterfaceAudience.Public  
  2. @InterfaceStability.Stable  
  3. public class CombineFileSplit extends InputSplit implements Writable {  
  4.   
  5.   private Path[] paths;  
  6.   private long[] startoffset;  
  7.   private long[] lengths;  
  8.   private String[] locations;  
  9.   private long totLength;  
  10.   
  11.   /** 
  12.    * default constructor 
  13.    */  
  14.   public CombineFileSplit() {}  
  15.   public CombineFileSplit(Path[] files, long[] start,   
  16.                           long[] lengths, String[] locations) {  
  17.     initSplit(files, start, lengths, locations);  
  18.   }  
  19.   
  20.   public CombineFileSplit(Path[] files, long[] lengths) {  
  21.     long[] startoffset = new long[files.length];  
  22.     for (int i = 0; i < startoffset.length; i++) {  
  23.       startoffset[i] = 0;  
  24.     }  
  25.     String[] locations = new String[files.length];  
  26.     for (int i = 0; i < locations.length; i++) {  
  27.       locations[i] = "";  
  28.     }  
  29.     initSplit(files, startoffset, lengths, locations);  
  30.   }  
  31.     
  32.   private void initSplit(Path[] files, long[] start,   
  33.                          long[] lengths, String[] locations) {  
  34.     this.startoffset = start;  
  35.     this.lengths = lengths;  
  36.     this.paths = files;  
  37.     this.totLength = 0;  
  38.     this.locations = locations;  
  39.     for(long length : lengths) {  
  40.       totLength += length;  
  41.     }  
  42.   }  
  43.   
  44.   /** 
  45.    * Copy constructor 
  46.    */  
  47.   public CombineFileSplit(CombineFileSplit old) throws IOException {  
  48.     this(old.getPaths(), old.getStartOffsets(),  
  49.          old.getLengths(), old.getLocations());  
  50.   }  
  51.   
  52.   public long getLength() {  
  53.     return totLength;  
  54.   }  
  55.   
  56.   /** Returns an array containing the start offsets of the files in the split*/   
  57.   public long[] getStartOffsets() {  
  58.     return startoffset;  
  59.   }  
  60.     
  61.   /** Returns an array containing the lengths of the files in the split*/   
  62.   public long[] getLengths() {  
  63.     return lengths;  
  64.   }  
  65.   
  66.   /** Returns the start offset of the i<sup>th</sup> Path */  
  67.   public long getOffset(int i) {  
  68.     return startoffset[i];  
  69.   }  
  70.     
  71.   /** Returns the length of the i<sup>th</sup> Path */  
  72.   public long getLength(int i) {  
  73.     return lengths[i];  
  74.   }  
  75.     
  76.   /** Returns the number of Paths in the split */  
  77.   public int getNumPaths() {  
  78.     return paths.length;  
  79.   }  
  80.   
  81.   /** Returns the i<sup>th</sup> Path */  
  82.   public Path getPath(int i) {  
  83.     return paths[i];  
  84.   }  
  85.     
  86.   /** Returns all the Paths in the split */  
  87.   public Path[] getPaths() {  
  88.     return paths;  
  89.   }  
  90.   
  91.   /** Returns all the Paths where this input-split resides */  
  92.   public String[] getLocations() throws IOException {  
  93.     return locations;  
  94.   }  
  95.   
  96.   public void readFields(DataInput in) throws IOException {  
  97.     totLength = in.readLong();  
  98.     int arrLength = in.readInt();  
  99.     lengths = new long[arrLength];  
  100.     for(int i=0; i<arrLength;i++) {  
  101.       lengths[i] = in.readLong();  
  102.     }  
  103.     int filesLength = in.readInt();  
  104.     paths = new Path[filesLength];  
  105.     for(int i=0; i<filesLength;i++) {  
  106.       paths[i] = new Path(Text.readString(in));  
  107.     }  
  108.     arrLength = in.readInt();  
  109.     startoffset = new long[arrLength];  
  110.     for(int i=0; i<arrLength;i++) {  
  111.       startoffset[i] = in.readLong();  
  112.     }  
  113.   }  
  114.   
  115.   public void write(DataOutput out) throws IOException {  
  116.     out.writeLong(totLength);  
  117.     out.writeInt(lengths.length);  
  118.     for(long length : lengths) {  
  119.       out.writeLong(length);  
  120.     }  
  121.     out.writeInt(paths.length);  
  122.     for(Path p : paths) {  
  123.       Text.writeString(out, p.toString());  
  124.     }  
  125.     out.writeInt(startoffset.length);  
  126.     for(long length : startoffset) {  
  127.       out.writeLong(length);  
  128.     }  
  129.   }  
  130.     
  131.   @Override  
  132.  public String toString() {  
  133.     StringBuffer sb = new StringBuffer();  
  134.     for (int i = 0; i < paths.length; i++) {  
  135.       if (i == 0 ) {  
  136.         sb.append("Paths:");  
  137.       }  
  138.       sb.append(paths[i].toUri().getPath() + ":" + startoffset[i] +  
  139.                 "+" + lengths[i]);  
  140.       if (i < paths.length -1) {  
  141.         sb.append(",");  
  142.       }  
  143.     }  
  144.     if (locations != null) {  
  145.       String locs = "";  
  146.       StringBuffer locsb = new StringBuffer();  
  147.       for (int i = 0; i < locations.length; i++) {  
  148.         locsb.append(locations[i] + ":");  
  149.       }  
  150.       locs = locsb.toString();  
  151.       sb.append(" Locations:" + locs + "; ");  
  152.     }  
  153.     return sb.toString();  
  154.   }  
  155. }  

与FileSPlit类似，CombineFileSplit同样包含文件路径，分片起始位置，分片大小和存储分片数据的host列表，由于CombineFileSplit是针对小文件的，它把很多小文件包在一个InputSplit中，这样一个Mapper就可以处理很多小文件。要知道我们上面的FileSplit是对应一个输入文件的也就是说如果用FileSplit对应的FileInputFormat来作为输入格式。那么即使文件特别小，也是单独计算成一个分片来处理的。当我们的输入是由大量小文件组成的，就会导致同样大量的InputSplit，从而需要同样大量的Mapper来处理，这将很慢，想想一堆Map Task要运行(运行一个新的MapTask可是要启动虚拟机的)，这是不符合Hadoop的设计理念的，所以使用CombineFileSplit可以优化Hadoop处理众多小文件的场景。

 

最后介绍TagInputSplit，这个类就是封装了一个InputSplit，然后加了一些tags在里面满足我们需要这些tags数据的情况，我们从下面就可以一目了然。

 

[java] view plain copy
 
  1. class TaggedInputSplit extends InputSplit implements Configurable, Writable {  
  2.   
  3.   private Class<? extends InputSplit> inputSplitClass;  
  4.   
  5.   private InputSplit inputSplit;  
  6.   
  7.   @SuppressWarnings("unchecked")  
  8.   private Class<? extends InputFormat> inputFormatClass;  
  9.   
  10.   @SuppressWarnings("unchecked")  
  11.   private Class<? extends Mapper> mapperClass;  
  12.   
  13.   private Configuration conf;  
  14.   
  15.   public TaggedInputSplit() {  
  16.     // Default constructor.  
  17.   }  
  18.   
  19.   /** 
  20.    * Creates a new TaggedInputSplit. 
  21.    *  
  22.    * @param inputSplit The InputSplit to be tagged 
  23.    * @param conf The configuration to use 
  24.    * @param inputFormatClass The InputFormat class to use for this job 
  25.    * @param mapperClass The Mapper class to use for this job 
  26.    */  
  27.   @SuppressWarnings("unchecked")  
  28.   public TaggedInputSplit(InputSplit inputSplit, Configuration conf,  
  29.       Class<? extends InputFormat> inputFormatClass,  
  30.       Class<? extends Mapper> mapperClass) {  
  31.     this.inputSplitClass = inputSplit.getClass();  
  32.     this.inputSplit = inputSplit;  
  33.     this.conf = conf;  
  34.     this.inputFormatClass = inputFormatClass;  
  35.     this.mapperClass = mapperClass;  
  36.   }  
  37.   
  38.   /** 
  39.    * Retrieves the original InputSplit. 
  40.    *  
  41.    * @return The InputSplit that was tagged 
  42.    */  
  43.   public InputSplit getInputSplit() {  
  44.     return inputSplit;  
  45.   }  
  46.   
  47.   /** 
  48.    * Retrieves the InputFormat class to use for this split. 
  49.    *  
  50.    * @return The InputFormat class to use 
  51.    */  
  52.   @SuppressWarnings("unchecked")  
  53.   public Class<? extends InputFormat> getInputFormatClass() {  
  54.     return inputFormatClass;  
  55.   }  
  56.   
  57.   /** 
  58.    * Retrieves the Mapper class to use for this split. 
  59.    *  
  60.    * @return The Mapper class to use 
  61.    */  
  62.   @SuppressWarnings("unchecked")  
  63.   public Class<? extends Mapper> getMapperClass() {  
  64.     return mapperClass;  
  65.   }  
  66.   
  67.   public long getLength() throws IOException, InterruptedException {  
  68.     return inputSplit.getLength();  
  69.   }  
  70.   
  71.   public String[] getLocations() throws IOException, InterruptedException {  
  72.     return inputSplit.getLocations();  
  73.   }  
  74.   
  75.   @SuppressWarnings("unchecked")  
  76.   public void readFields(DataInput in) throws IOException {  
  77.     inputSplitClass = (Class<? extends InputSplit>) readClass(in);  
  78.     inputFormatClass = (Class<? extends InputFormat<?, ?>>) readClass(in);  
  79.     mapperClass = (Class<? extends Mapper<?, ?, ?, ?>>) readClass(in);  
  80.     inputSplit = (InputSplit) ReflectionUtils  
  81.        .newInstance(inputSplitClass, conf);  
  82.     SerializationFactory factory = new SerializationFactory(conf);  
  83.     Deserializer deserializer = factory.getDeserializer(inputSplitClass);  
  84.     deserializer.open((DataInputStream)in);  
  85.     inputSplit = (InputSplit)deserializer.deserialize(inputSplit);  
  86.   }  
  87.   
  88.   private Class<?> readClass(DataInput in) throws IOException {  
  89.     String className = Text.readString(in);  
  90.     try {  
  91.       return conf.getClassByName(className);  
  92.     } catch (ClassNotFoundException e) {  
  93.       throw new RuntimeException("readObject can‘t find class", e);  
  94.     }  
  95.   }  
  96.   
  97.   @SuppressWarnings("unchecked")  
  98.   public void write(DataOutput out) throws IOException {  
  99.     Text.writeString(out, inputSplitClass.getName());  
  100.     Text.writeString(out, inputFormatClass.getName());  
  101.     Text.writeString(out, mapperClass.getName());  
  102.     SerializationFactory factory = new SerializationFactory(conf);  
  103.     Serializer serializer =   
  104.           factory.getSerializer(inputSplitClass);  
  105.     serializer.open((DataOutputStream)out);  
  106.     serializer.serialize(inputSplit);  
  107.   }  
  108.   
  109.   public Configuration getConf() {  
  110.     return conf;  
  111.   }  
  112.   
  113.   public void setConf(Configuration conf) {  
  114.     this.conf = conf;  
  115.   }  
  116.   
  117. }  


InputFormat：

 

通过使用InputFormat，MapReduce框架可以做到：

1.验证作业输入的正确性。

2.将输入文件切分成逻辑的InputSplits，一个InputSplit将被分配给一个单独的MapTask。

3.提供RecordReader的实现，这个RecordReader会从InputSplit中正确读出一条一条的K-V对供Mapper使用。

 

[java] view plain copy
 
  1. public abstract class InputFormat<K, V> {  
  2.   
  3.    
  4.   public abstract   
  5.     List<InputSplit> getSplits(JobContext context  
  6.                                ) throws IOException, InterruptedException;  
  7.     
  8.    
  9.   public abstract   
  10.     RecordReader<K,V> createRecordReader(InputSplit split,  
  11.                                          TaskAttemptContext context  
  12.                                         ) throws IOException,   
  13.                                                  InterruptedException;  
  14.   
  15. }  

上面是InputFormat的源码，getSplits()是用来获取由输入文件计算出来的InputSplits，我们在后面会看到计算InputSplit的时候会考虑到输入文件是否可分割、文件存储时分块的大小和文件大小等因素；而createRecordReader()提供了前面第三点所说的RecordReader的实现，以将K-V对从InputSplit中正确读取出来，比如LineRecordReader就以偏移值为key，一行数据为value的形式读取分片的。

 

 

FileInputFormat：

PathFilter被用来进行文件刷选，这样我们就可以控制哪些文件要被作为输入，哪些不作为输入，PathFIlter有一个accept(Path)方法，当接收的Path要被包含进来，就返回true，否则返回false。可以通过设置mapred.input.pathFIlter.class来设置用户自定义的PathFilter。

 

[java] view plain copy
 
  1. public interface PathFilter {  
  2.   /** 
  3.    * Tests whether or not the specified abstract pathname should be 
  4.    * included in a pathname list. 
  5.    * 
  6.    * @param  path  The abstract pathname to be tested 
  7.    * @return  <code>true</code> if and only if <code>pathname</code> 
  8.    *          should be included 
  9.    */  
  10.   boolean accept(Path path);  
  11. }  

FileInputFormat是InputFormat的子类，它包含了一个MultiPathFilter，这个MultiPathFilter由一个过滤隐藏文件(名字前缀‘-‘或‘.‘)的PathFilter和一些可能存在的用户自定义的PathFilter组成，MultiPathFilter会在listStatus()方法中使用，而listStatus()方法又被getSplits()方法用来获取输入文件，也就是说实现了在获取输入分片前进行文件过滤。

 

 

[java] view plain copy
 
  1. private static class MultiPathFilter implements PathFilter {  
  2.    private List<PathFilter> filters;  
  3.   
  4.    public MultiPathFilter() {  
  5.      this.filters = new ArrayList<PathFilter>();  
  6.    }  
  7.   
  8.    public MultiPathFilter(List<PathFilter> filters) {  
  9.      this.filters = filters;  
  10.    }  
  11.   
  12.    public void add(PathFilter one) {  
  13.      filters.add(one);  
  14.    }  
  15.   
  16.    public boolean accept(Path path) {  
  17.      for (PathFilter filter : filters) {  
  18.        if (filter.accept(path)) {  
  19.          return true;  
  20.        }  
  21.      }  
  22.      return false;  
  23.    }  
  24.   
  25.    public String toString() {  
  26.      StringBuffer buf = new StringBuffer();  
  27.      buf.append("[");  
  28.      for (PathFilter f: filters) {  
  29.        buf.append(f);  
  30.        buf.append(",");  
  31.      }  
  32.      buf.append("]");  
  33.      return buf.toString();  
  34.    }  
  35.  }  

这些PathFilter会在listStatus()方法中用到，listStatus()是用来获取输入数据列表的。

 

下面是FileInputFormat的getSplits()方法，它首先得到分片的最小值minSize和最大值maxSize，它们会被用来计算分片的大小。可以通过设置mapred.min.split.size和mapred.max.split.size来设置。splits集合可以用来存储计算得到的输入分片，files则存储作为由listStatus()获取的输入文件列表。然后对于每个输入文件，判断是否可以分割，通过computeSplits()计算出分片大小splitSize，计算方法是：Math.max(minSize,Math.min(maxSize,blockSize));也就是保证在minSize和maxSize之间，且如果minSize<=blockSize<=maxSize，则设blockSize。然后根据这个splitSize计算出每个文件的InputSplit集合，然后加入到列表splits集合中。注意到我们生成InputSplit的时候按上面说的使用文件路径，分片起始位置，分片大小和存放这个文件爱你的hosts列表来创建。最后我们还设置了输入文件数量：mapreduce.input.num.files。

 

[java] view plain copy
 
  1. public List<InputSplit> getSplits(JobContext job  
  2.                                    ) throws IOException {  
  3.    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  
  4.    long maxSize = getMaxSplitSize(job);  
  5.   
  6.    // generate splits  
  7.    List<InputSplit> splits = new ArrayList<InputSplit>();  
  8.    List<FileStatus>files = listStatus(job);  
  9.    for (FileStatus file: files) {  
  10.      Path path = file.getPath();  
  11.      FileSystem fs = path.getFileSystem(job.getConfiguration());  
  12.      long length = file.getLen();  
  13.      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);  
  14.      if ((length != 0) && isSplitable(job, path)) {   
  15.        long blockSize = file.getBlockSize();  
  16.        long splitSize = computeSplitSize(blockSize, minSize, maxSize);  
  17.   
  18.        long bytesRemaining = length;  
  19.        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  
  20.          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);  
  21.          splits.add(new FileSplit(path, length-bytesRemaining, splitSize,   
  22.                                   blkLocations[blkIndex].getHosts()));  
  23.          bytesRemaining -= splitSize;  
  24.        }  
  25.          
  26.        if (bytesRemaining != 0) {  
  27.          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,   
  28.                     blkLocations[blkLocations.length-1].getHosts()));  
  29.        }  
  30.      } else if (length != 0) {  
  31.        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));  
  32.      } else {   
  33.        //Create empty hosts array for zero length files  
  34.        splits.add(new FileSplit(path, 0, length, new String[0]));  
  35.      }  
  36.    }  
  37.      
  38.    // Save the number of input files in the job-conf  
  39.    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());  
  40.   
  41.    LOG.debug("Total # of splits: " + splits.size());  
  42.    return splits;  
  43.  }  

就这样，我们利用FileInputFormat的getSplits()方法，我们就计算出了我们作业的所有输入分片了。

 

那这些计算出来的分片是怎么被map读出来的呢？就是InputFormat中的另一个方法createRecordReader()，FileInputFormat并没有对这个方法做具体的要求，而是交给子类自行去实现它。

 

RecordReader：

RecordReader是用来从一个输入分片中读取一个一个的K-V对的抽象类，我们可以将其看做是在InputSplit上的迭代器。我们从类图中可以看到它的一些方法，最主要的方法就是nextKeyValue()方法，由它获取分片上的下一个K-V对。

我们呢再来看看RecordReader的一个子类：LineRecordReader，这也是我们用的最多的。

LineRecordReader由一个FileSplit构造出来，start是这个FileSplit的起始位置，pos是当前读取分片的位置，end是分片结束位置，in是打开一个读取这个分片的输入流，它是使用这个FIleSplit对应的文件名来打开的。key和value则分别是每次读取的K-V对。然后我们还看到可以利用getProgress()来跟踪读取分片的进度，这个函数就是根据已经读取的K-V对占总K-V对的比例显示进度的。

 

[java] view plain copy
 
  1. public class LineRecordReader extends RecordReader<LongWritable, Text> {  
  2.   private static final Log LOG = LogFactory.getLog(LineRecordReader.class);  
  3.   
  4.   private CompressionCodecFactory compressionCodecs = null;  
  5.   private long start;  
  6.   private long pos;  
  7.   private long end;  
  8.   private LineReader in;  
  9.   private int maxLineLength;  
  10.   private LongWritable key = null;  
  11.   private Text value = null;  
  12.   private Seekable filePosition;  
  13.   private CompressionCodec codec;  
  14.   private Decompressor decompressor;  
  15.   
  16.   public void initialize(InputSplit genericSplit,  
  17.                          TaskAttemptContext context) throws IOException {  
  18.     FileSplit split = (FileSplit) genericSplit;  
  19.     Configuration job = context.getConfiguration();  
  20.     this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",  
  21.                                     Integer.MAX_VALUE);  
  22.     start = split.getStart();  
  23.     end = start + split.getLength();  
  24.     final Path file = split.getPath();  
  25.     compressionCodecs = new CompressionCodecFactory(job);  
  26.     codec = compressionCodecs.getCodec(file);  
  27.   
  28.     // open the file and seek to the start of the split  
  29.     FileSystem fs = file.getFileSystem(job);  
  30.     FSDataInputStream fileIn = fs.open(split.getPath());  
  31.   
  32.     if (isCompressedInput()) {  
  33.       decompressor = CodecPool.getDecompressor(codec);  
  34.       if (codec instanceof SplittableCompressionCodec) {  
  35.         final SplitCompressionInputStream cIn =  
  36.           ((SplittableCompressionCodec)codec).createInputStream(  
  37.             fileIn, decompressor, start, end,  
  38.             SplittableCompressionCodec.READ_MODE.BYBLOCK);  
  39.         in = new LineReader(cIn, job);  
  40.         start = cIn.getAdjustedStart();  
  41.         end = cIn.getAdjustedEnd();  
  42.         filePosition = cIn;  
  43.       } else {  
  44.         in = new LineReader(codec.createInputStream(fileIn, decompressor),  
  45.             job);  
  46.         filePosition = fileIn;  
  47.       }  
  48.     } else {  
  49.       fileIn.seek(start);  
  50.       in = new LineReader(fileIn, job);  
  51.       filePosition = fileIn;  
  52.     }  
  53.     // If this is not the first split, we always throw away first record  
  54.     // because we always (except the last split) read one extra line in  
  55.     // next() method.  
  56.     if (start != 0) {  
  57.       start += in.readLine(new Text(), 0, maxBytesToConsume(start));  
  58.     }  
  59.     this.pos = start;  
  60.   }  
  61.     
  62.   private boolean isCompressedInput() {  
  63.     return (codec != null);  
  64.   }  
  65.   
  66.   private int maxBytesToConsume(long pos) {  
  67.     return isCompressedInput()  
  68.       ? Integer.MAX_VALUE  
  69.       : (int) Math.min(Integer.MAX_VALUE, end - pos);  
  70.   }  
  71.   
  72.   private long getFilePosition() throws IOException {  
  73.     long retVal;  
  74.     if (isCompressedInput() && null != filePosition) {  
  75.       retVal = filePosition.getPos();  
  76.     } else {  
  77.       retVal = pos;  
  78.     }  
  79.     return retVal;  
  80.   }  
  81.   
  82.   public boolean nextKeyValue() throws IOException {  
  83.     if (key == null) {  
  84.       key = new LongWritable();  
  85.     }  
  86.     key.set(pos);  
  87.     if (value == null) {  
  88.       value = new Text();  
  89.     }  
  90.     int newSize = 0;  
  91.     // We always read one extra line, which lies outside the upper  
  92.     // split limit i.e. (end - 1)  
  93.     while (getFilePosition() <= end) {  
  94.       newSize = in.readLine(value, maxLineLength,  
  95.           Math.max(maxBytesToConsume(pos), maxLineLength));  
  96.       if (newSize == 0) {  
  97.         break;  
  98.       }  
  99.       pos += newSize;  
  100.       if (newSize < maxLineLength) {  
  101.         break;  
  102.       }  
  103.   
  104.       // line too long. try again  
  105.       LOG.info("Skipped line of size " + newSize + " at pos " +   
  106.                (pos - newSize));  
  107.     }  
  108.     if (newSize == 0) {  
  109.       key = null;  
  110.       value = null;  
  111.       return false;  
  112.     } else {  
  113.       return true;  
  114.     }  
  115.   }  
  116.   
  117.   @Override  
  118.   public LongWritable getCurrentKey() {  
  119.     return key;  
  120.   }  
  121.   
  122.   @Override  
  123.   public Text getCurrentValue() {  
  124.     return value;  
  125.   }  
  126.   
  127.   /** 
  128.    * Get the progress within the split 
  129.    */  
  130.   public float getProgress() throws IOException {  
  131.     if (start == end) {  
  132.       return 0.0f;  
  133.     } else {  
  134.       return Math.min(1.0f,  
  135.         (getFilePosition() - start) / (float)(end - start));  
  136.     }  
  137.   }  
  138.   
  139.   public synchronized void close() throws IOException {  
  140.     try {  
  141.       if (in != null) {  
  142.         in.close();  
  143.       }  
  144.     } finally {  
  145.       if (decompressor != null) {  
  146.         CodecPool.returnDecompressor(decompressor);  
  147.       }  
  148.     }  
  149.   }  
  150. }  

其它的一些RecordReader如SequenceFileRecordReader，CombineFileRecordReader则对应不同的InputFormat。

 

 

下面继续看看这些RecordReader是如何被MapReduce框架使用的。

首先看看Mapper类的源码：

 

[java] view plain copy
 
  1. public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {  
  2.   
  3.   public class Context   
  4.     extends MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {  
  5.     public Context(Configuration conf, TaskAttemptID taskid,  
  6.                    RecordReader<KEYIN,VALUEIN> reader,  
  7.                    RecordWriter<KEYOUT,VALUEOUT> writer,  
  8.                    OutputCommitter committer,  
  9.                    StatusReporter reporter,  
  10.                    InputSplit split) throws IOException, InterruptedException {  
  11.       super(conf, taskid, reader, writer, committer, reporter, split);  
  12.     }  
  13.   }  
  14.     
  15.   /** 
  16.    * Called once at the beginning of the task. 
  17.    */  
  18.   protected void setup(Context context  
  19.                        ) throws IOException, InterruptedException {  
  20.     // NOTHING  
  21.   }  
  22.   
  23.   /** 
  24.    * Called once for each key/value pair in the input split. Most applications 
  25.    * should override this, but the default is the identity function. 
  26.    */  
  27.   @SuppressWarnings("unchecked")  
  28.   protected void map(KEYIN key, VALUEIN value,   
  29.                      Context context) throws IOException, InterruptedException {  
  30.     context.write((KEYOUT) key, (VALUEOUT) value);  
  31.   }  
  32.   
  33.   /** 
  34.    * Called once at the end of the task. 
  35.    */  
  36.   protected void cleanup(Context context  
  37.                          ) throws IOException, InterruptedException {  
  38.     // NOTHING  
  39.   }  
  40.     
  41.   /** 
  42.    * Expert users can override this method for more complete control over the 
  43.    * execution of the Mapper. 
  44.    * @param context 
  45.    * @throws IOException 
  46.    */  
  47.   public void run(Context context) throws IOException, InterruptedException {  
  48.     setup(context);  
  49.     while (context.nextKeyValue()) {  
  50.       map(context.getCurrentKey(), context.getCurrentValue(), context);  
  51.     }  
  52.     cleanup(context);  
  53.   }  
  54. }  

我们写MapReduce程序的时候，我们写的Mapper类都要继承这个Mapper类，通常我们会重写map()方法，map()每次接受一个K-V对，然后对这个K-V对进行处理，再分发出处理后的数据。我们也可能重写setUp()方法以对这个MapTask进行一些预处理，比如创建一个List之类集合，我们也可能重写cleanUp()方法做一些处理后的工作，当然我们也可能在cleanUp()中写出K-V对。举个例子就是：InputSplit的数据是一些整数，然后我们要在Mapper中计算他们的和。我们可以先设置个sum属性，然后map()函数处理一个K-V对就是将其加到sum上，最后在cleanUp()函数中调用context.write(key,value)。

 

最后我们看看Mapper.class中的run()方法，它相当于MapTask的驱动，我们可以看到run()方法首先调用setUp()方法进行初始化操作，然后遍历每个通过context.nextKeyValue()获取的K-V对，调用map()函数进行处理，最后调用cleanUp()方法做相关处理。

 

我们看看Mapper.class中的Context类，它继承自MapContext，使用了一个RecordReader进行构造。下面我们看看MapContext这个类的源码：

 

[java] view plain copy
 
  1. public class MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskInputOutputContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {  
  2.     private RecordReader<KEYIN, VALUEIN> reader;  
  3.     private InputSplit split;  
  4.   
  5.     public MapContext(Configuration conf, TaskAttemptID taskid, RecordReader<KEYIN, VALUEIN> reader, RecordWriter<KEYOUT, VALUEOUT> writer,  
  6.             OutputCommitter committer, StatusReporter reporter, InputSplit split) {  
  7.         super(conf, taskid, writer, committer, reporter);  
  8.         this.reader = reader;  
  9.         this.split = split;  
  10.     }  
  11.   
  12.     /** 
  13.      * Get the input split for this map. 
  14.      */  
  15.     public InputSplit getInputSplit() {  
  16.         return split;  
  17.     }  
  18.   
  19.     @Override  
  20.     public KEYIN getCurrentKey() throws IOException, InterruptedException {  
  21.         return reader.getCurrentKey();  
  22.     }  
  23.   
  24.     @Override  
  25.     public VALUEIN getCurrentValue() throws IOException, InterruptedException {  
  26.         return reader.getCurrentValue();  
  27.     }  
  28.   
  29.     @Override  
  30.     public boolean nextKeyValue() throws IOException, InterruptedException {  
  31.         return reader.nextKeyValue();  
  32.     }  
  33.   
  34. }  

我们可以看到MapContext直接是使用传入的RecordReader来进行K-V对的读取。

 

 

到现在，我们已经知道输入文件是如何被读取、过滤、分片、读出K-V对，然后交给我们的Mapper类来处理的了。

 

最后，我们来看看FIleInputFormat的几个子类。

TextInputFormat：

TextInputFormat是FileInputFormat的子类，其createRecordReader()方法返回的就是LineRecordReader。

 

[java] view plain copy
 
  1. public class TextInputFormat extends FileInputFormat<LongWritable, Text> {  
  2.   
  3.   @Override  
  4.   public RecordReader<LongWritable, Text>   
  5.     createRecordReader(InputSplit split,  
  6.                        TaskAttemptContext context) {  
  7.     return new LineRecordReader();  
  8.   }  
  9.   
  10.   @Override  
  11.   protected boolean isSplitable(JobContext context, Path file) {  
  12.     CompressionCodec codec =   
  13.       new CompressionCodecFactory(context.getConfiguration()).getCodec(file);  
  14.     if (null == codec) {  
  15.       return true;  
  16.     }  
  17.     return codec instanceof SplittableCompressionCodec;  
  18.   }  
  19.   
  20. }  

我们还看到isSplitable()方法，当文件使用压缩的形式，这个文件就不可分割，否则就读取不正确的数据了。这从某种程度上将影响分片的计算。有时我们希望一个文件只被一个Mapper处理的时候，我们就可以重写isSplitable()方法，告诉MapReduce框架，我哪些文件可以分割，哪些文件不能分割而只能作为一个分片。

 

 

NLineInputFormat：

NLineInputFormat也是FileInputFormat的子类，与名字一致，它是根据行数来划分InputSplits而不是像TextInputFormat那样依赖分片大小和行的长度的。也就是说，TextInputFormat当一行很长或分片比较小时，获取的分片可能只包含很少的K-V对，这样一个MapTask处理的K-V对就很少，这可能很不理想。因此我们可以使用NLineInputFormat来控制一个MapTask处理的K-V对，这是通过分割InputSplit时，按行数来分割的方法来实现的，这我们在代码中可以看出来。我们设置mapreduce.input.lineinputformat.linespermap来设置这个行数，源码如下：

 

[java] view plain copy
 
    1. @InterfaceAudience.Public  
    2. @InterfaceStability.Stable  
    3. public class NLineInputFormat extends FileInputFormat<LongWritable, Text> {   
    4.   public static final String LINES_PER_MAP =   
    5.     "mapreduce.input.lineinputformat.linespermap";  
    6.   
    7.   public RecordReader<LongWritable, Text> createRecordReader(  
    8.       InputSplit genericSplit, TaskAttemptContext context)   
    9.       throws IOException {  
    10.     context.setStatus(genericSplit.toString());  
    11.     return new LineRecordReader();  
    12.   }  
    13.   
    14.   /**  
    15.    * Logically splits the set of input files for the job, splits N lines 
    16.    * of the input as one split. 
    17.    *  
    18.    * @see FileInputFormat#getSplits(JobContext) 
    19.    */  
    20.   public List<InputSplit> getSplits(JobContext job)  
    21.   throws IOException {  
    22.     List<InputSplit> splits = new ArrayList<InputSplit>();  
    23.     int numLinesPerSplit = getNumLinesPerSplit(job);  
    24.     for (FileStatus status : listStatus(job)) {  
    25.       splits.addAll(getSplitsForFile(status,  
    26.         job.getConfiguration(), numLinesPerSplit));  
    27.     }  
    28.     return splits;  
    29.   }  
    30.     
    31.   public static List<FileSplit> getSplitsForFile(FileStatus status,  
    32.       Configuration conf, int numLinesPerSplit) throws IOException {  
    33.     List<FileSplit> splits = new ArrayList<FileSplit> ();  
    34.     Path fileName = status.getPath();  
    35.     if (status.isDir()) {  
    36.       throw new IOException("Not a file: " + fileName);  
    37.     }  
    38.     FileSystem  fs = fileName.getFileSystem(conf);  
    39.     LineReader lr = null;  
    40.     try {  
    41.       FSDataInputStream in  = fs.open(fileName);  
    42.       lr = new LineReader(in, conf);  
    43.       Text line = new Text();  
    44.       int numLines = 0;  
    45.       long begin = 0;  
    46.       long length = 0;  
    47.       int num = -1;  
    48.       while ((num = lr.readLine(line)) > 0) {  
    49.         numLines++;  
    50.         length += num;  
    51.         if (numLines == numLinesPerSplit) {  
    52.           splits.add(createFileSplit(fileName, begin, length));  
    53.           begin += length;  
    54.           length = 0;  
    55.           numLines = 0;  
    56.         }  
    57.       }  
    58.       if (numLines != 0) {  
    59.         splits.add(createFileSplit(fileName, begin, length));  
    60.       }  
    61.     } finally {  
    62.       if (lr != null) {  
    63.         lr.close();  
    64.       }  
    65.     }  
    66.     return splits;   
    67.   }  
    68.   
    69.   /** 
    70.    * NLineInputFormat uses LineRecordReader, which always reads 
    71.    * (and consumes) at least one character out of its upper split 
    72.    * boundary. So to make sure that each mapper gets N lines, we 
    73.    * move back the upper split limits of each split  
    74.    * by one character here. 
    75.    * @param fileName  Path of file 
    76.    * @param begin  the position of the first byte in the file to process 
    77.    * @param length  number of bytes in InputSplit 
    78.    * @return  FileSplit 
    79.    */  
    80.   protected static FileSplit createFileSplit(Path fileName, long begin, long length) {  
    81.     return (begin == 0)   
    82.     ? new FileSplit(fileName, begin, length - 1, new String[] {})  
    83.     : new FileSplit(fileName, begin - 1, length, new String[] {});  
    84.   }  
    85.     
    86.   /** 
    87.    * Set the number of lines per split 
    88.    * @param job the job to modify 
    89.    * @param numLines the number of lines per split 
    90.    */  
    91.   public static void setNumLinesPerSplit(Job job, int numLines) {  
    92.     job.getConfiguration().setInt(LINES_PER_MAP, numLines);  
    93.   }  
    94.   
    95.   /** 
    96.    * Get the number of lines per split 
    97.    * @param job the job 
    98.    * @return the number of lines per split 
    99.    */  
    100.   public static int getNumLinesPerSplit(JobContext job) {  
    101.     return job.getConfiguration().getInt(LINES_PER_MAP, 1);  
    102.   }  
    103. }  

Hadoop InputFormat源码分析

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173722.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！