hadoop输入分片计算(Map Task个数的确定)

时间：2014-11-21 23:04:51 阅读：250 评论：0 收藏：0 [点我收藏+]

　　作业从JobClient端的submitJobInternal()方法提交作业的同时，调用InputFormat接口的getSplits()方法来创建split。默认是使用InputFormat的子类FileInputFormat来计算分片，而split的默认实现为FileSplit(其父接口为InputSplit)。这里要注意，split只是逻辑上的概念，并不对文件做实际的切分。一个split记录了一个Map Task要处理的文件区间，所以分片要记录其对应的文件偏移量以及长度等。每个split由一个Map Task来处理，所以有多少split，就有多少Map Task。下面着重分析这个方法：

 1 public List<InputSplit> getSplits(JobContext job
 2                                     ) throws IOException {
 3     //getFormatMinSplitSize()：始终返回1
 4     //getMinSplitSize(job)：获取” mapred.min.split.size”的值，默认为1
 5     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
 6     
 7     //getMaxSplitSize(job)：获取"mapred.max.split.size"的值，
 8     //默认配置文件中并没有这一项，所以其默认值为” Long.MAX_VALUE”，即2^63 – 1
 9     long maxSize = getMaxSplitSize(job);
10 
11     // generate splits
12     List<InputSplit> splits = new ArrayList<InputSplit>();
13     List<FileStatus>files = listStatus(job);
14     for (FileStatus file: files) {
15       Path path = file.getPath();
16       FileSystem fs = path.getFileSystem(job.getConfiguration());
17       long length = file.getLen();
18       BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
19       if ((length != 0) && isSplitable(job, path)) { 
20         long blockSize = file.getBlockSize();
21         //计算split大小
22         long splitSize = computeSplitSize(blockSize, minSize, maxSize);
23 
24         //计算split个数
25         long bytesRemaining = length;    //bytesRemaining表示剩余字节数
26         while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { //SPLIT_SLOP=1.1
27           int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
28           splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
29                                    blkLocations[blkIndex].getHosts()));
30           bytesRemaining -= splitSize;
31         }
32         
33         if (bytesRemaining != 0) {
34           splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
35                      blkLocations[blkLocations.length-1].getHosts()));
36         }
37       } else if (length != 0) {
38         splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
39       } else { 
40         //Create empty hosts array for zero length files
41         splits.add(new FileSplit(path, 0, length, new String[0]));
42       }
43     }
44     
45     // Save the number of input files in the job-conf
46     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
47 
48     LOG.debug("Total # of splits: " + splits.size());
49     return splits;
50   }

　　首先计算分片的下限和上限：minSize和maxSize，具体的过程在注释中已经说清楚了。接下来用这两个值再加上blockSize来计算实际的split大小，过程也很简单，具体代码如下：

1 protected long computeSplitSize(long blockSize, long minSize,
2                                   long maxSize) {
3     return Math.max(minSize, Math.min(maxSize, blockSize));
4   }