在前面一篇文章中（hadoop2.7之作业提交详解（上））中涉及到文件的分片。
JobSubmitter.submitJobInternal方法中调用了
int maps = writeSplits(job,submitJobDir); //设置map的数量，而map的数量是根据文件的大小和分片的大小，以及文件的数量决定的
接下来我们看一下JobSubmitter.writeSplits方法：
@H_301_10@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方法:
@H_301_10@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这个抽象类：
@H_301_10@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类只重写了两个方法：如下：
@H_301_10@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()方法：
@H_301_10@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的个数。
