一、MapReduce的执行大致流程

1、InputFormat数据读取流程

1.1、切片

  1.   - 切片的概念：从文件的逻辑上进行大小的切分，一个切片多大，将来一个MapTask的处理的数据就多大
  2.   - 一个切片就会产生一个MapTask
  3.   - 切片时只考虑文件本身，不考虑数据的整体集
  4.   - 切片大小和切块大小默认是一致的，这样设计的目的是为了避免将来切片读取数据的时候有跨机器的情况

1.2、InputFormat的体系结构

  1. -- FileInputFormat  InputFormat的子实现类，实现切片逻辑
  2.              实现了getSplits() 负责切片
  4. -- TextInputFormat  FileInputFormat的子实现类， 实现读取数据的逻辑
  5.              createRecordReader() 返回一个RecordReader，在RecordReader中实现了
  6.              读取数据的方式：安行读取。
  8. -- CombineFileInputFormat  FileInputFormat的子实现类，此类中也实现了
  9.              一套切片逻辑 （处理：适用于小文件计算场景。）
  11. // 切片源码分析
  12.         public List<InputSplit> getSplits(JobContext job) throws IOException {
  13.             StopWatch sw = new StopWatch().start();
  15.             // minSize = 1(默认情况) 
  16.                // 但是我们也可以通过改变mapreduce.input.fileinputformat.split.minsize 配置项来改变minSize大小
  18.             long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
  20.             // maxSize = Long的最大值(默认情况)
  21.             // 但是我们也可以通过改变mapreduce.input.fileinputformat.split.maxsize 配置项来改变maxSize大小
  22.             long maxSize = getMaxSplitSize(job);
  24.             // 管理最终切完片的对象的集合 最终返回的就是此集合
  25.             List<InputSplit> splits = new ArrayList<InputSplit>();
  27.             // 获取当前文件的详情
  28.             List<FileStatus> files = listStatus(job);
  30.             boolean ignoreDirs = !getInputDirRecursive(job)
  31.               && job.getConfiguration().getBoolean(INPUT_DIR_NONRECURSIVE_IGNORE_SUBDIRS, false);
  33.             // 遍历获取到的文件列表，一次按照文件为单位进行切片  
  34.             for (FileStatus file: files) {
  36.               // 如果是忽略文件以及是文件夹就不进行切片
  37.               if (ignoreDirs && file.isDirectory()) {
  38.                 continue;
  39.               }
  41.               // 获取文件的路径
  42.               Path path = file.getPath();
  43.               // 获取文件的内容大小
  44.               long length = file.getLen();
  45.               // 如果不是空文件 继续切片
  46.               if (length != 0) {
  48.                 // 获取文件的具体的块信息
  49.                 BlockLocation[] blkLocations;
  50.                 if (file instanceof LocatedFileStatus) {
  51.                   blkLocations = ((LocatedFileStatus) file).getBlockLocations();
  52.                 } else {
  53.                   FileSystem fs = path.getFileSystem(job.getConfiguration());
  54.                   blkLocations = fs.getFileBlockLocations(file, 0, length);
  55.                 }
  57.                 // 判断是否要进行切片（主要判断当前文件是否是压缩文件，有一些压缩文件时不能够进行切片）
  58.                 if (isSplitable(job, path)) {
  59.                   // 获取HDFS中的数据块的大小
  60.                   long blockSize = file.getBlockSize();
  61.                   // 计算切片的大小--> 128M 默认情况下永远都是块大小
  62.                   long splitSize = computeSplitSize(blockSize, minSize, maxSize);
  63.                   -- 内部方法：
  64.                       protected long computeSplitSize(long blockSize, long minSize,
  65.                                   long maxSize) {
  66.                             return Math.max(minSize, Math.min(maxSize, blockSize));
  67.                          }
  69.                   long bytesRemaining = length;
  71.                   // 判断当前的文件的剩余内容是否要继续切片 SPLIT_SLOP = 1.1
  72.                   // 判断公式：bytesRemaining)/splitSize > SPLIT_SLOP
  73.                   // 用文件的剩余大小/切片大小 > 1.1 才继续切片（这样做的目的是为了让我们每一个MapTask处理的数据更加均衡）
  74.                   while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  75.                     int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  76.                     splits.add(makeSplit(path, length-bytesRemaining, splitSize,
  77.                                 blkLocations[blkIndex].getHosts(),
  78.                                 blkLocations[blkIndex].getCachedHosts()));
  79.                     bytesRemaining -= splitSize;
  80.                   }
  82.                   // 如果最后文件还有剩余且不足一个切片大小，最后再形成最后的一个切片
  83.                   if (bytesRemaining != 0) {
  84.                     int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  85.                     splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
  86.                                blkLocations[blkIndex].getHosts(),
  87.                                blkLocations[blkIndex].getCachedHosts()));
  88.                   }
  89.                 } else { // not splitable
  90.                   if (LOG.isDebugEnabled()) {
  91.                     // Log only if the file is big enough to be splitted
  92.                     if (length > Math.min(file.getBlockSize(), minSize)) {
  93.                       LOG.debug("File is not splittable so no parallelization "
  94.                           + "is possible: " + file.getPath());
  95.                     }
  96.                   }
  97.                   splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
  98.                               blkLocations[0].getCachedHosts()));
  99.                 }
  100.               } else { 
  101.                 //Create empty hosts array for zero length files
  102.                 splits.add(makeSplit(path, 0, length, new String[0]));
  103.               }
  104.             }
  105.             // Save the number of input files for metrics/loadgen
  106.             job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
  107.             sw.stop();
  108.             if (LOG.isDebugEnabled()) {
  109.               LOG.debug("Total # of splits generated by getSplits: " + splits.size()
  110.                   + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
  111.             }
  112.             return splits;
  113.           }

2、Shuffle工作机制