Hadoop中Map的数量是如何确定的

        我们知道，在Hadoop中作业运行的时候，Map的数量是由输入分片的数量决定的，但是分片的数量，并不是简单的按照文件的大小和blockSize的大小来切分的，分片的数量其实也是经过一系列的计算得到的，我们常用的InputFormat很多都是继承自FileInputFormat，该类时个抽象类，但是其中的getSplits方法是有完整的实现的，如果我们没有刻意去覆盖的话，也是会按照该方法中的逻辑来得到分片的，下面对该方法中的代码进行分析：

  public List<InputSplit> getSplits(JobContext job) throws IOException {
    //得到分片的最小值
    //其中getFormatMinSplitSize()默认返回值为1
    //getMinSplitSize(job)会返回mapred.min.split.size的值，默认值为0
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    //该值不做设置的话会返回Long.MAX_VALUE
    long maxSize = getMaxSplitSize(job);

    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<FileStatus>files = listStatus(job);
    //对每个文件做分片
    for (FileStatus file: files) {
      Path path = file.getPath();
      FileSystem fs = path.getFileSystem(job.getConfiguration());
      //文件的大小
      long length = file.getLen();
      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
      //如果文件大小不为0，并且文件时可分割的
      if ((length != 0) && isSplitable(job, path)) {
        //blockSize 默认为64M
        long blockSize = file.getBlockSize();
        /*computeSplitSize()的源码：
          Math.max(minSize, Math.min(maxSize, blockSize))
          也就是说默认的splitSize=blockSize
        */
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);
        
        long bytesRemaining = length;
        //注意这里的条件,并不是大小超过splitSize就一定会切分，还需要超过splitSize 10%，
        //这里的SPLIT_SLOP值为1.1
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
                                   blkLocations[blkIndex].getHosts()));
          bytesRemaining -= splitSize;
        }
        
        //经过上面的循环切分后，剩下的部分，单独作为一个切片
        if (bytesRemaining != 0) {
          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
                     blkLocations[blkLocations.length-1].getHosts()));
        }
      } else if (length != 0) {//如果文件不支持切分
        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
      } else { //如果文件大小为0
        //Create empty hosts array for zero length files
        splits.add(new FileSplit(path, 0, length, new String[0]));
      }
    }
    
    // Save the number of input files in the job-conf
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());

    LOG.debug("Total # of splits: " + splits.size());
    return splits;
  }