在前面一篇文章中(hadoop2.7之作业提交详解(上))中涉及到文件的分片。
JobSubmitter.submitJobInternal方法中调用了
int maps = writeSplits(job, submitJobDir); //设置map的数量,而map的数量是根据文件的大小和分片的大小,以及文件的数量决定的
接下来我们看一下JobSubmitter.writeSplits方法:
private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,
Path jobSubmitDir) throws IOException,
InterruptedException, ClassNotFoundException {
JobConf jConf = (JobConf)job.getConfiguration();
int maps;
if (jConf.getUseNewMapper()) {
maps = writeNewSplits(job, jobSubmitDir); //这里我们使用新的方式
} else {
maps = writeOldSplits(jConf, jobSubmitDir);
}
return maps;
}
接下来继续看JobSubmitter.writeNewSplits方法:
private <T extends InputSplit>
int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
InterruptedException, ClassNotFoundException {
Configuration conf = job.getConfiguration();
InputFormat<?, ?> input =
ReflectionUtils.newInstance(job.getInputFormatClass(), conf); //输入对象,InputFormat是个抽象类 List<InputSplit> splits = input.getSplits(job); //调用InputFormat实现类的getSplits方法
T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]); // sort the splits into order based on size, so that the biggest
// go first
Arrays.sort(array, new SplitComparator()); //对切片的大小进行排序,最大的放最前面
JobSplitWriter.createSplitFiles(jobSubmitDir, conf,
jobSubmitDir.getFileSystem(conf), array);//创建Split文件
return array.length;
}
接下来看一下InputFormat这个抽象类:
public abstract class InputFormat<K, V> {
//用来返回分片结果
public abstract
List<InputSplit> getSplits(JobContext context
) throws IOException, InterruptedException;
//RecordReader是用来从一个输入分片中读取一个一个的K-V对的抽象类,我们可以将其看作是在InputSplit上的迭代器。
//最主要的方法就是nextKeyvalue()方法,由它获取分片上的下一个K-V 对。
public abstract
RecordReader<K,V> createRecordReader(InputSplit split,
TaskAttemptContext context
) throws IOException,
InterruptedException; }
接下来我们继续看这个抽象类的实现类:
public class TextInputFormat extends FileInputFormat;
public abstract class FileInputFormat<K, V> extends InputFormat;
public abstract class InputFormat。
由于TextInputFormat从抽象类FileInputFormat中继承,所以大部分的方法都来自于FileInputFormat类,TextInputFormat类只重写了两个方法:如下:
public class TextInputFormat extends FileInputFormat<LongWritable, Text> { @Override
public RecordReader<LongWritable, Text>
createRecordReader(InputSplit split,
TaskAttemptContext context) {
String delimiter = context.getConfiguration().get(
"textinputformat.record.delimiter");
byte[] recordDelimiterBytes = null;
if (null != delimiter)
recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
//LineRecordReader由一个FileSplit构造出来,start是这个FileSplit的起始位置,pos是当前读取分片的位置,
//end是分片结束位置,in是打开的一个读取这个分片的输入流,它是使用这个FileSplit对应的文件名来打开的。
//key和value则分别是每次读取的K-V对。然后我们还看到可以利用getProgress()来跟踪读取分片的进度,
//这个函数就是根据已经读取的K-V对占总K-V对的比例来显示进度的
return new LineRecordReader(recordDelimiterBytes);
} @Override
protected boolean isSplitable(JobContext context, Path file) {
//如果是压缩文件就不切分,非压缩文件就切分。
final CompressionCodec codec =
new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
if (null == codec) {
return true;
}
return codec instanceof SplittableCompressionCodec;
}
}
我们在返回到JobSubmitter.writeNewSplits方法中,有List<InputSplit> splits = input.getSplits(job);主要是调用了TextInputFormat.getSplits()方法,而TextInputFormat继承了FileInputFormat类,所以调用的就是FileInputFormat.getSplits()方法:
public List<InputSplit> getSplits(JobContext job) throws IOException {
StopWatch sw = new StopWatch().start();//用来计算纳秒级别的时间
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); //最小值默认为1
long maxSize = getMaxSplitSize(job); //最大值为long的最大值,默认为0x7fffffffffffffffL // generate splits
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;
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)) { //判断是否可以分片
long blockSize = file.getBlockSize(); //128M
long splitSize = computeSplitSize(blockSize, minSize, maxSize); //计算分片的大小,默认为128M long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { //判断剩余文件大小是否大于128M*1.1
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);//f返回每个分片起始位置
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
bytesRemaining -= splitSize; // 依次减去分片的大小,对剩余长度再次分片
}
// 多次分片后,最后的数据长度仍不为0但又不足一个分片大小
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()); //设置参数NUM_INPUT_FILES
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
}
return splits;
}
//public class FileSplit extends InputSplit implements Writable {
// private Path file;//输入文件路径
// private long start;//分片在文件中的位置(起点)
// private long length;//分片长度
// private String[] hosts;//这个分片所在数据块的多个复份所在节点
// private SplitLocationInfo[] hostInfos;//每个数据块复份所在节点,以及是否缓存
//}
//makeSplit方法存放的分片格式
protected FileSplit makeSplit(Path file, long start, long length,
String[] hosts, String[] inMemoryHosts) {
return new FileSplit(file, start, length, hosts, inMemoryHosts);
} //计算分片的大小
protected long computeSplitSize(long blockSize, long minSize,
long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
通过FileInputFormat.getSplits(),可以返回一个存放分片的ArraryList,接下继续回到JobSubmitter.writeNewSplits方法中:
接下来将ArrayList转换为数组,并根据分片的大小排序。然后调用JobSplitWriter.createSplitFiles()方法创建split文件。最后返回数组的长度,也就是map的个数。