public InputSplit[] getSplits(JobConf job, int numSplits)

//每个DN的最小split大小 
long minSizeNode = 0;
//同机架的最小split大小 
long minSizeRack = 0;
//最大的split大小 
long maxSize = 0;

/** 

     * Return all the splits in the specified set of paths 

     *  

* @param job Job的配置信息 

* @param paths 输入源的路径列表 

* @param maxSize 最大的split大小 

* @param minSizeNode 每个DN最小的split大小 

* @param minSizeRack 每个rack最小的split大小 

* @param splits split信息列表 

* @throws IOException 

*/
private void getMoreSplits(JobConf job, Path[] paths,
long maxSize, long minSizeNode, long minSizeRack,
List<CombineFileSplit> splits)

// populate all the blocks for all files
long totLength = 0;
for (int i = 0; i < paths.length; i++) {
      //构建每个input文件的信息，并将文件中的每个 
      //block信息收集到rackToBlocks、blockToNodes、nodeToBlocks中 
      files[i] = new OneFileInfo(paths[i], job,
      rackToBlocks, blockToNodes, nodeToBlocks);
      //增加所有文件的大小 
      totLength += files[i].getLength();
}

// process all nodes and create splits that are local
// to a node.
//创建同一个DN上的split 
for (Iterator<Map.Entry<String,
List<OneBlockInfo>>> iter = nodeToBlocks.entrySet().iterator();
iter.hasNext();) {
      Map.Entry<String, List<OneBlockInfo>> one = iter.next();
      nodes.add(one.getKey());
      List<OneBlockInfo> blocksInNode = one.getValue();
      // for each block, copy it into validBlocks. Delete it from
      // blockToNodes so that the same block does not appear in
      // two different splits.
      for (OneBlockInfo oneblock : blocksInNode) {
            if (blockToNodes.containsKey(oneblock)) {
                  validBlocks.add(oneblock);
                  blockToNodes.remove(oneblock);
                  curSplitSize += oneblock.length;
                  // if the accumulated split size exceeds the maximum, then
                  // create this split.
                  if (maxSize != 0 && curSplitSize >= maxSize) {
                        // create an input split and add it to the splits array
                        //创建这些block合并后的split，并将其split添加到split列表中 
                        addCreatedSplit(job, splits, nodes, validBlocks);
                        //重置 
                        curSplitSize = 0;
                        validBlocks.clear();
                  }
            }
      }
      // if there were any blocks left over and their combined size is
      // larger than minSplitNode, then combine them into one split.
      // Otherwise add them back to the unprocessed pool. It is likely
      // that they will be combined with other blocks from the same rack later on.
      //其实这里的注释已经说的很清楚，我再按照我的理解��一下 
      /** 

          * 这里有几种情况： 

          * 1、在这个DN上还有没有被split的block， 

          * 而且这些block的大小大于了在一个DN上的split最小值（没有达到最大值）， 

          * 将把这些block合并成一个split 

          * 2、剩余的block的大小还是没有达到，将剩余的这些block 

          * 归还给blockToNodes，等以后统一处理 

          */
      if (minSizeNode != 0 && curSplitSize >= minSizeNode) {
            // create an input split and add it to the splits array
            addCreatedSplit(job, splits, nodes, validBlocks);
      } else {
      for (OneBlockInfo oneblock : validBlocks) {
            blockToNodes.put(oneblock, oneblock.hosts);
      }
}
validBlocks.clear();
nodes.clear();
curSplitSize = 0;
}

// if blocks in a rack are below the specified minimum size, then keep them
// in 'overflow'. After the processing of all racks is complete, these overflow
// blocks will be combined into splits.
ArrayList<OneBlockInfo> overflowBlocks = new ArrayList<OneBlockInfo>();
ArrayList<String> racks = new ArrayList<String>();
// Process all racks over and over again until there is no more work to do.
//这里处理的就不再是同一个DN上的block 
//同一个DN上的已经被处理过了（上面的代码），这里是一些 
//还没有被处理的block 
while (blockToNodes.size() > 0) {
      // Create one split for this rack before moving over to the next rack.
      // Come back to this rack after creating a single split for each of the
      // remaining racks.
      // Process one rack location at a time, Combine all possible blocks that
      // reside on this rack as one split. (constrained by minimum and maximum
      // split size).
      // iterate over all racks
      //创建同机架的split 
      for (Iterator<Map.Entry<String, List<OneBlockInfo>>> iter =
      rackToBlocks.entrySet().iterator(); iter.hasNext();) {
            Map.Entry<String, List<OneBlockInfo>> one = iter.next();
            racks.add(one.getKey());
            List<OneBlockInfo> blocks = one.getValue();
            // for each block, copy it into validBlocks. Delete it from
            // blockToNodes so that the same block does not appear in
            // two different splits.
            boolean createdSplit = false;
            for (OneBlockInfo oneblock : blocks) {
                  //这里很重要，现在的blockToNodes说明的是还有哪些block没有被split 
                  if (blockToNodes.containsKey(oneblock)) {
                        validBlocks.add(oneblock);
                        blockToNodes.remove(oneblock);
                        curSplitSize += oneblock.length;
                        // if the accumulated split size exceeds the maximum, then
                        // create this split.
                        if (maxSize != 0 && curSplitSize >= maxSize) {
                              // create an input split and add it to the splits array
                              addCreatedSplit(job, splits, getHosts(racks), validBlocks);
                              createdSplit = true;
                              break;
                        }
                  }
            }
            // if we created a split, then just go to the next rack
            if (createdSplit) {
                  curSplitSize = 0;
                  validBlocks.clear();
                  racks.clear();
                  continue;
            }
            //还有没有被split的block 
            //如果这些block的大小大于了同机架的最小split， 
            //则创建split 
            //否则，将这些block留到后面处理 
            if (!validBlocks.isEmpty()) {
                  if (minSizeRack != 0 && curSplitSize >= minSizeRack) {
                        // if there is a mimimum size specified, then create a single split
                        // otherwise, store these blocks into overflow data structure
                        addCreatedSplit(job, splits, getHosts(racks), validBlocks);
                  } else {
                  // There were a few blocks in this rack that remained to be processed.
                  // Keep them in 'overflow' block list. These will be combined later.
                  overflowBlocks.addAll(validBlocks);
            }
      }
      curSplitSize = 0;
      validBlocks.clear();
      racks.clear();
}
}

MultiFileCombineInputFormat
package org.rollinkin.hadoop;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.CombineFileInputFormat;
import org.apache.hadoop.mapred.lib.CombineFileRecordReader;
import org.apache.hadoop.mapred.lib.CombineFileSplit;
/** 

    * 多文件合并split的输入format 

    *  

    * @author rollinkin 

    * @date 2012-10-29 

    * @version 1.0 

    * @since 1.0 

    */
public class MultiFileCombineInputFormat extends
CombineFileInputFormat<LongWritable, Text> {
      @Override
      public RecordReader<LongWritable, Text> getRecordReader(
      InputSplit split, JobConf job, Reporter reporter)
      throws IOException {
      @SuppressWarnings({ "rawtypes", "unchecked" })
            Class<RecordReader<LongWritable, Text>> rrClass = (Class)CombineLineRecordReader.class;
            return new CombineFileRecordReader<LongWritable, Text>(job,(CombineFileSplit) split, reporter,rrClass);
      }
}

package org.rollinkin.hadoop;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.CombineFileSplit;
public class CombineLineRecordReader implements
RecordReader<LongWritable, Text> {
      private LineRecordReader delegate;
      public CombineLineRecordReader(CombineFileSplit split, Configuration conf,
      Reporter reporter, Integer idx) throws IOException {
            FileSplit fileSplit = new FileSplit(split.getPath(idx),
            split.getOffset(idx), split.getLength(idx),
            split.getLocations());
            delegate = new LineRecordReader(conf, fileSplit);
      }
      @Override
      public boolean next(LongWritable key, Text value) throws IOException {
            return delegate.next(key, value);
      }
      @Override
      public LongWritable createKey() {
            return delegate.createKey();
      }
      @Override
      public Text createValue() {
            return delegate.createValue();
      }
      @Override
      public long getPos() throws IOException {
            return delegate.getPos();
      }
      @Override
      public void close() throws IOException {
            delegate.close();
      }
      @Override
      public float getProgress() throws IOException {
            return delegate.getProgress();
      }
}