1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
public InputSplit[] getSplits(JobConf job, int numSplits)
throws IOException {
// 得到输入文件的各种状态
FileStatus[] files = listStatus(job);
// Save the number of input files in the job-conf
// conf 中设置输入文件的数目
job.setLong(NUM_INPUT_FILES, files.length);
// 计算总的大小
long totalSize = 0; // compute total size
for (FileStatus file: files) { // check we have valid files
if (file.isDir()) {
throw new IOException("Not a file: "+ file.getPath());
}
totalSize += file.getLen();
}
// numSplits 传进来的是 map 的数目
// 获得每一个分片的期望大小
long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
// 获得最小的分片大小,这个可以在 mapred.min.split.size 中设置
long minSize = Math.max(job.getLong("mapred.min.split.size", 1),
minSplitSize);
// generate splits
// 以下是生成分片的计算
ArrayList<FileSplit> splits = new ArrayList<FileSplit>(numSplits);
NetworkTopology clusterMap = new NetworkTopology();
for (FileStatus file: files) {
Path path = file.getPath();
FileSystem fs = path.getFileSystem(job);
long length = file.getLen();
BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
// isSplitable 是判断该文件是否可以分片
// 一般情况下都是可以的,但是如果是 stream compressed 的方式,那么是不可以的
if ((length != 0) && isSplitable(fs, path)) {
long blockSize = file.getBlockSize();
// 计算每一个分片大小的实际函数
// 得到真实的分片大小
long splitSize = computeSplitSize(goalSize, minSize, blockSize);
long bytesRemaining = length;
// 允许最后一个分片在 SPLIT_SLOP(默认 1.1) 比例之下
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[] splitHosts = getSplitHosts(blkLocations,
length-bytesRemaining, splitSize, clusterMap);
// 加入分片
splits.add(new FileSplit(path, length-bytesRemaining, splitSize,
splitHosts));
bytesRemaining -= splitSize;
}
// 加入最后一个分片
// 这个比例最大不超过期望分片的 1.1
if (bytesRemaining != 0) {
splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkLocations.length-1].getHosts()));
}
} else if (length != 0) {
String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
splits.add(new FileSplit(path, 0, length, splitHosts));
} else {
//Create empty hosts array for zero length files
splits.add(new FileSplit(path, 0, length, new String[0]));
}
}
LOG.debug("Total # of splits: " + splits.size());
return splits.toArray(new FileSplit[splits.size()]);
}
protected long computeSplitSize(long goalSize, long minSize,
long blockSize) {
// 计算分片大小,很明显
// 这里设定了最大最小值,每一个分片大小在 minSize 和 blockSize 之间
return Math.max(minSize, Math.min(goalSize, blockSize));
}
|