在前面一篇文章中（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()来跟踪读取分片的进度， 这个函数就是根据已经读取的K-V对占总K-V对的比例来显示进度的 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_FILES sw.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的个数。
原文链接：/bigdata/997880.html
