1. 定义
1.1 block
block是物理块，文件存放到HDFS上后，会将大文件按照每块128MB的大小切分，存放到不同的DataNode上。
1.2 split
split是逻辑上的分片，在MapReduce中Map开始之前，会将输入文件按照指定大小切分为多个小片，每一部分对应一个Map Task，默认split的大小与block的大小相同，为128MB。
2. 参数设置
2.1 block默认配置在hdfs-site.xml中【$HADOOP_HOME/share/hadoop/hdfs/hadoop-hdfs-x.y.z.jar】
<property>
<name>dfs.blocksize</name>
<value>134217728</value>
<description>
默认block的大小参数配置以字节为单位（例如134217728，128 MB）。
也可以使用如128k，512m，1g等为单位（不区分大小写）。
</description>
</property>
2.2 split大小由minSize、maxSize、blockSize决定
- long minSize = 1
在FileInputFormat.getSplits方法中，minSize的赋值：// 取getFormatMinSplitSize()、getMinSplitSize(job)的最大值 long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); protected long getFormatMinSplitSize() { return 1; } public static long getMinSplitSize(JobContext job) { return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L); } - long maxSize = Long.MAX_VALUE
在FileInputFormat.getSplits方法中，maxSize的赋值：long maxSize = getMaxSplitSize(job); public static long getMaxSplitSize(JobContext context) { return context.getConfiguration().getLong(SPLIT_MAXSIZE, Long.MAX_VALUE); } public long getLong(String name, long defaultValue) { // 用户自定义了参数 String valueString = getTrimmed(name); if (valueString == null) // 返回默认值Long.MAX_VALUE return defaultValue; String hexString = getHexDigits(valueString); if (hexString != null) { return Long.parseLong(hexString, 16); } return Long.parseLong(valueString); } - long blockSize = 128MB
2.3 结论
在FileInputFormat.getSplits方法中对文件进行了Split：
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
protected long computeSplitSize(long blockSize, long minSize, long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
image.png
split与block的对应关系可以是多对一，默认一对一：
- 如果blockSize < maxSize && blockSize < minSize，那么split就是blockSize【一对一】
- 如果blockSize < maxSize && blockSize > minSize，那么split就是minSize
- 如果blockSize > maxSize && maxSize > minSize，那么split就是maxSize【多对一】
如果blockSize > maxSize && maxSize < minSize，那么split就是minSize
3. 分片主要源码
public List<InputSplit> getSplits(JobContext job) throws IOException {
Stopwatch sw = new Stopwatch().start();
// 分片最小值，默认为 1
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
// 分片最大值，默认为 LONG.MAX_VALUE
long maxSize = getMaxSplitSize(job);
// 生成分片列表
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) {
BlockLocation[] blkLocations;
// 获得block块的位置信息
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
FileSystem fs = path.getFileSystem(job.getConfiguration());
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
// 判断文件是否可以分片
if (isSplitable(job, path)) {
// 获得文件blockSize大小，默认128MB
long blockSize = file.getBlockSize();
// 获得split大小，默认128MB
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
// 文件剩余大小
long bytesRemaining = length;
// 当剩余大小大于split大小的1.1倍时，进行分片
// private static final double SPLIT_SLOP = 1.1;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
// 获得block块的索引位置
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
// 分片
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
// 原文件减去已经分片大小
bytesRemaining -= splitSize;
}
// 判断文件是否已经完成分片，如果还有剩余，则将剩余部分作为一个分片
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
}
} else { // not splitable
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
blkLocations[0].getCachedHosts()));
}
} else {
//Create empty hosts array for zero length 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());
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.elapsedMillis());
}
return splits;
}
// 分片，封装分片信息，文件路径，分片起始位置，分片大小，对应block块所在的hosts列表，对应block块缓存副本所在的hosts列表
protected FileSplit makeSplit(Path file, long start, long length, String[] hosts, String[] inMemoryHosts) {
return new FileSplit(file, start, length, hosts, inMemoryHosts);
}
写在最后
上面说到的，当剩余大小大于split大小的1.1倍时，进行分片
private static final double SPLIT_SLOP = 1.1;
我还没有想出问什么是1.1倍，我猜想是为了减少一些分片数量，比如这种情况？
这种
