在前面一篇文章中(hadoop2.7之做业提交详解(上))中涉及到文件的分片。html
JobSubmitter.submitJobInternal方法中调用了
int maps = writeSplits(job, submitJobDir); //设置map的数量,而map的数量是根据文件的大小和分片的大小,以及文件的数量决定的sql
接下来咱们看一下JobSubmitter.writeSplits方法:apache
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; }
接下来咱们继续看这个抽象类的实现类:app
public class TextInputFormat extends FileInputFormat;
public abstract class FileInputFormat<K, V> extends InputFormat;
public abstract class InputFormat。ide
因为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()方法:oop
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方法中:post
接下来将ArrayList转换为数组,并根据分片的大小排序。而后调用JobSplitWriter.createSplitFiles()方法建立split文件。最后返回数组的长度,也就是map的个数。