Hadoop InputFormat源码分析

Posted 2020-06-12

平时我们写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^th Path */  
67.   public long getOffset(int i) {  
68.     return startoffset[i];  
69.   }  
70.     
71.   /** Returns the length of the i^th 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^th 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.