在前面一篇文章中(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();
- 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) 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文件
- array.length;
- }
接下来看一下InputFormat这个抽象类:
- public abstract class InputFormat<K,V> {
- 用来返回分片结果
- abstract
- List<InputSplit> getSplits(JobContext context
- ) RecordReader是用来从一个输入分片中读取一个一个的K-V对的抽象类,我们可以将其看作是在InputSplit上的迭代器。
- 最主要的方法就是nextKeyvalue()方法,由它获取分片上的下一个K-V 对。
- RecordReader<K,1)"> createRecordReader(InputSplit split,TaskAttemptContext context
- ) 方法都来自于FileInputFormat类,TextInputFormat类只重写了两个方法:如下:
class TextInputFormat extends FileInputFormat<LongWritable,Text> {@Override
public RecordReader<LongWritable,1)"> 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()来跟踪读取分片的进度, return LineRecordReader(recordDelimiterBytes);}
@Override
protected boolean isSplitable(JobContext context,Path file) { final CompressionCodec codec = CompressionCodecFactory(context.getConfiguration()).getCodec(file); null == codec) { 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) 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) { bytesRemaining); splits.add(makeSplit(path,length-不可分,则把整个文件作为一个分片 } not splitable splits.add(makeSplit(path,length,blkLocations[0].getHosts(),blkLocations[0].getCachedHosts()));}
} {创建空的分片
Create empty hosts array for zero length files splits.add(makeSplit(path,1)">new String[0]));}
}
Save the number of input files for metrics/loadgen job.getConfiguration().setLong(NUM_INPUT_FILES,files.size()); 设置参数NUM_INPUT_FILESsw.stop();
(LOG.isDebugEnabled()) { LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ",TiMetaken: " + sw.now(TimeUnit.MILLISECONDS));}
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,1)">long start,1)">long length,String[] hosts,String[] inMemoryHosts) { FileSplit(file,start,hosts,inMemoryHosts);}
计算分片的大小
long computeSplitSize(long blockSize,1)"> minSize, maxSize) { Math.max(minSize,Math.min(maxSize,blockSize));}
通过FileInputFormat.getSplits(),可以返回一个存放分片的ArraryList,接下继续回到JobSubmitter.writeNewSplits方法中:
接下来将ArrayList转换为数组,并根据分片的大小排序。然后调用JobSplitWriter.createSplitFiles()方法创建split文件。最后返回数组的长度,也就是map的个数。