1.常见输入格式

1.1 TextInputFormat 默认的切片策略
RecordReader: LineRecordReader,一次处理一行，将一行内容的偏移量作为key，一行内容作为value!
LongWritable key
Text value
1.2NLineInputFormat
RecordReader: LineRecordReader,几行生成一个切片 一次处理一行，将一行内容的偏移量作为key，一行内容作为value!
LongWritable key
Text value
使用场景：当行逻辑复杂执行时间长
1.3 KeyValueTextInputFormat
RecordReader：KeyValueLineRecordReader 针对文本文件！使用分割字符，将每一行分割为key和value!
如果没有找到分隔符，当前行的内容作为key，value为空串!
默认分隔符为\t，可以通过参数mapreduce.input.keyvaluelinerecordreader.key.value.separator指定！（分割符byte类 型只能一个字符）
切片：默认的切片策略
RR ： KeyValueLineRecordReader
Text key:
Text value
1.4ConbineTextInputFormat
作用： 改变了传统的切片方式！将多个小文件，划分到一个切片中！
适合小文件过多的场景！
RecordReader: LineRecordReader,一次处理一行，将一行内容的偏移量作为key，一行内容作为value!
LongWritable key
Text value
切片： 先确定片的最大值maxSize，maxSize通过参数mapreduce.input.fileinputformat.split.maxsize设置！
流程：a. 以文件为单位，将每个文件划分为若干part
①判断文件的待切部分的大小<= maxSize,整个待切部分作为1part
②maxsize < 文件的待切部分的大小 <= 2 maxSize,将整个待切部分均分为2part
③文件的待切部分的大小> 2 maxSize,先切去maxSize大小，作为1部分，剩余待切部分继续

3.提交流程分析

1.断电入口 job.waitForCompletion(true);
2.执行submit

  1.  public void submit() 
  2.          throws IOException, InterruptedException, ClassNotFoundException {
  3.     ensureState(JobState.DEFINE);
  4.      //1.检查是否是最新的api
  5.     setUseNewAPI();
  6.     connect();
  8.     final JobSubmitter submitter =  getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
  9.     status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
  10.       public JobStatus run() throws IOException, InterruptedException, 
  11.       ClassNotFoundException {
  12.            //2.执行提交流程
  13.         return submitter.submitJobInternal(Job.this, cluster);
  14.       }
  15.     });
  16.    }

3.submit分析

  1.     #计算切片
  2.   int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
  3.       InterruptedException, ClassNotFoundException {
  4.     Configuration conf = job.getConfiguration();
  5.      #根据driver配置的inputFormatClass参数反序列化出输入格式
  6.     InputFormat<?, ?> input =
  7.       ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
  8.     #此处调用
  9.     List<InputSplit> splits = input.getSplits(job);
  10.     T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);

4.核心逻辑

  1. public List<InputSplit> getSplits(JobContext job) throws IOException {
  2.     #getFormatMinSplitSize=1,config里面取配置mapreduce.input.fileinputformat.split.minsize配置参数默认0
  3.     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));#结果1
  4.     #config 取mapreduce.input.fileinputformat.split.maxsize 如果没有设置的话是long的最大值
  5.     long maxSize = getMaxSplitSize(job);
  7.     // generate splits
  8.     List<InputSplit> splits = new ArrayList<InputSplit>();
  9.     //查询所有文件的源数据信息
  10.     List<FileStatus> files = listStatus(job);
  11.     for (FileStatus file: files) {
  12.       Path path = file.getPath();
  13.       long length = file.getLen();
  14.       if (length != 0) {
  15.         BlockLocation[] blkLocations;
  16.         if (file instanceof LocatedFileStatus) {
  17.           blkLocations = ((LocatedFileStatus) file).getBlockLocations();
  18.         } else {
  19.           FileSystem fs = path.getFileSystem(job.getConfiguration());
  20.           blkLocations = fs.getFileBlockLocations(file, 0, length);
  21.         }
  22.         //查看该文本格式是否支持切割，不同文本格式不同
  23.         if (isSplitable(job, path)) {
  24.           long blockSize = file.getBlockSize();
  25.          /**
  26.          * protected long computeSplitSize(long blockSize, long minSize,long maxSize) {
  27.          如果将块调大，那么将minsSize调大
  28.          如果将快调小,那么将maxSize跳下，小于blockSize
  29.          * return Math.max(minSize, Math.min(maxSize, blockSize));
  30.          * }
  31.          */
  32.           long splitSize = computeSplitSize(blockSize, minSize, maxSize);
  34.           long bytesRemaining = length;
  35.           //切割到剩余1.1时不再切割
  36.           while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  37.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  38.             splits.add(makeSplit(path, length-bytesRemaining, splitSize,
  39.                         blkLocations[blkIndex].getHosts(),
  40.                         blkLocations[blkIndex].getCachedHosts()));
  41.             bytesRemaining -= splitSize;
  42.           }
  43.         //将剩余的作为一个切片
  44.           if (bytesRemaining != 0) {
  45.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  46.             splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
  47.                        blkLocations[blkIndex].getHosts(),
  48.                        blkLocations[blkIndex].getCachedHosts()));
  49.           }
  50.         } else { // not splitable 不能切割的话直接放到一个切片里
  51.           splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
  52.                       blkLocations[0].getCachedHosts()));
  53.         }
  54.       } else { 
  55.         //Create empty hosts array for zero length files
  56.         splits.add(makeSplit(path, 0, length, new String[0]));
  57.       }
  58.     }
  59.     return splits;

切片大小的调节 :
如果将块调大，那么将minsSize调大 mapreduce.input.fileinputformat.split.minsize
如果将快调小,那么将maxSize跳下，小于blockSize mapreduce.input.fileinputformat.split.maxsize