底层是使用CombineFileInputFormat的createSplits

    1.   void createSplits(Map<String, Set<OneBlockInfo>> nodeToBlocks,
    2.                      Map<OneBlockInfo, String[]> blockToNodes,
    3.                      Map<String, List<OneBlockInfo>> rackToBlocks,
    4.                      long totLength,
    5.                      long maxSize,
    6.                      long minSizeNode,
    7.                      long minSizeRack,
    8.                      List<InputSplit> splits                     
    9.                     ) {
    10.     ArrayList<OneBlockInfo> validBlocks = new ArrayList<OneBlockInfo>();
    11.     long curSplitSize = 0;
    13.     int totalNodes = nodeToBlocks.size();
    14.     long totalLength = totLength;
    16.     Multiset<String> splitsPerNode = HashMultiset.create();
    17.     Set<String> completedNodes = new HashSet<String>();
    19.     while(true) {
    20.       // it is allowed for maxSize to be 0. Disable smoothing load for such cases
    22.       // process all nodes and create splits that are local to a node. Generate
    23.       // one split per node iteration, and walk over nodes multiple times to
    24.       // distribute the splits across nodes. 
    25.       for (Iterator<Map.Entry<String, Set<OneBlockInfo>>> iter = nodeToBlocks
    26.           .entrySet().iterator(); iter.hasNext();) {
    27.         Map.Entry<String, Set<OneBlockInfo>> one = iter.next();
    29.         String node = one.getKey();
    31.         // Skip the node if it has previously been marked as completed.
    32.         if (completedNodes.contains(node)) {
    33.           continue;
    34.         }
    36.         Set<OneBlockInfo> blocksInCurrentNode = one.getValue();
    38.         // for each block, copy it into validBlocks. Delete it from
    39.         // blockToNodes so that the same block does not appear in
    40.         // two different splits.
    41.         Iterator<OneBlockInfo> oneBlockIter = blocksInCurrentNode.iterator();
    42.         while (oneBlockIter.hasNext()) {
    43.           OneBlockInfo oneblock = oneBlockIter.next();
    45.           // Remove all blocks which may already have been assigned to other
    46.           // splits.
    47.           if(!blockToNodes.containsKey(oneblock)) {
    48.             oneBlockIter.remove();
    49.             continue;
    50.           }
    52.           validBlocks.add(oneblock);
    53.           blockToNodes.remove(oneblock);
    54.           curSplitSize += oneblock.length;
    56.           // if the accumulated split size exceeds the maximum, then
    57.           // create this split.
    58.           if (maxSize != 0 && curSplitSize >= maxSize) {
    59.             // create an input split and add it to the splits array
    60.             addCreatedSplit(splits, Collections.singleton(node), validBlocks);
    61.             totalLength -= curSplitSize;
    62.             curSplitSize = 0;
    64.             splitsPerNode.add(node);
    66.             // Remove entries from blocksInNode so that we don't walk these
    67.             // again.
    68.             blocksInCurrentNode.removeAll(validBlocks);
    69.             validBlocks.clear();
    71.             // Done creating a single split for this node. Move on to the next
    72.             // node so that splits are distributed across nodes.
    73.             break;
    74.           }
    76.         }
    77.         if (validBlocks.size() != 0) {
    78.           // This implies that the last few blocks (or all in case maxSize=0)
    79.           // were not part of a split. The node is complete.
    81.           // if there were any blocks left over and their combined size is
    82.           // larger than minSplitNode, then combine them into one split.
    83.           // Otherwise add them back to the unprocessed pool. It is likely
    84.           // that they will be combined with other blocks from the
    85.           // same rack later on.
    86.           // This condition also kicks in when max split size is not set. All
    87.           // blocks on a node will be grouped together into a single split.
    88.           if (minSizeNode != 0 && curSplitSize >= minSizeNode
    89.               && splitsPerNode.count(node) == 0) {
    90.             // haven't created any split on this machine. so its ok to add a
    91.             // smaller one for parallelism. Otherwise group it in the rack for
    92.             // balanced size create an input split and add it to the splits
    93.             // array
    94.             addCreatedSplit(splits, Collections.singleton(node), validBlocks);
    95.             totalLength -= curSplitSize;
    96.             splitsPerNode.add(node);
    97.             // Remove entries from blocksInNode so that we don't walk this again.
    98.             blocksInCurrentNode.removeAll(validBlocks);
    99.             // The node is done. This was the last set of blocks for this node.
    100.           } else {
    101.             // Put the unplaced blocks back into the pool for later rack-allocation.
    102.             for (OneBlockInfo oneblock : validBlocks) {
    103.               blockToNodes.put(oneblock, oneblock.hosts);
    104.             }
    105.           }
    106.           validBlocks.clear();
    107.           curSplitSize = 0;
    108.           completedNodes.add(node);
    109.         } else { // No in-flight blocks.
    110.           if (blocksInCurrentNode.size() == 0) {
    111.             // Node is done. All blocks were fit into node-local splits.
    112.             completedNodes.add(node);
    113.           } // else Run through the node again.
    114.         }
    115.       }
    117.       // Check if node-local assignments are complete.
    118.       if (completedNodes.size() == totalNodes || totalLength == 0) {
    119.         // All nodes have been walked over and marked as completed or all blocks
    120.         // have been assigned. The rest should be handled via rackLock assignment.
    121.         LOG.info("DEBUG: Terminated node allocation with : CompletedNodes: "
    122.             + completedNodes.size() + ", size left: " + totalLength);
    123.         break;
    124.       }
    125.     }
    127.     // if blocks in a rack are below the specified minimum size, then keep them
    128.     // in 'overflow'. After the processing of all racks is complete, these 
    129.     // overflow blocks will be combined into splits.
    130.     ArrayList<OneBlockInfo> overflowBlocks = new ArrayList<OneBlockInfo>();
    131.     Set<String> racks = new HashSet<String>();
    133.     // Process all racks over and over again until there is no more work to do.
    134.     while (blockToNodes.size() > 0) {
    136.       // Create one split for this rack before moving over to the next rack. 
    137.       // Come back to this rack after creating a single split for each of the 
    138.       // remaining racks.
    139.       // Process one rack location at a time, Combine all possible blocks that
    140.       // reside on this rack as one split. (constrained by minimum and maximum
    141.       // split size).
    143.       // iterate over all racks 
    144.       for (Iterator<Map.Entry<String, List<OneBlockInfo>>> iter = 
    145.            rackToBlocks.entrySet().iterator(); iter.hasNext();) {
    147.         Map.Entry<String, List<OneBlockInfo>> one = iter.next();
    148.         racks.add(one.getKey());
    149.         List<OneBlockInfo> blocks = one.getValue();
    151.         // for each block, copy it into validBlocks. Delete it from 
    152.         // blockToNodes so that the same block does not appear in 
    153.         // two different splits.
    154.         boolean createdSplit = false;
    155.         for (OneBlockInfo oneblock : blocks) {
    156.           if (blockToNodes.containsKey(oneblock)) {
    157.             validBlocks.add(oneblock);
    158.             blockToNodes.remove(oneblock);
    159.             curSplitSize += oneblock.length;
    161.             // if the accumulated split size exceeds the maximum, then 
    162.             // create this split.
    163.             if (maxSize != 0 && curSplitSize >= maxSize) {
    164.               // create an input split and add it to the splits array
    165.               addCreatedSplit(splits, getHosts(racks), validBlocks);
    166.               createdSplit = true;
    167.               break;
    168.             }
    169.           }
    170.         }
    172.         // if we created a split, then just go to the next rack
    173.         if (createdSplit) {
    174.           curSplitSize = 0;
    175.           validBlocks.clear();
    176.           racks.clear();
    177.           continue;
    178.         }
    180.         if (!validBlocks.isEmpty()) {
    181.           if (minSizeRack != 0 && curSplitSize >= minSizeRack) {
    182.             // if there is a minimum size specified, then create a single split
    183.             // otherwise, store these blocks into overflow data structure
    184.             addCreatedSplit(splits, getHosts(racks), validBlocks);
    185.           } else {
    186.             // There were a few blocks in this rack that 
    187.             // remained to be processed. Keep them in 'overflow' block list. 
    188.             // These will be combined later.
    189.             overflowBlocks.addAll(validBlocks);
    190.           }
    191.         }
    192.         curSplitSize = 0;
    193.         validBlocks.clear();
    194.         racks.clear();
    195.       }
    196.     }
    198.     assert blockToNodes.isEmpty();
    199.     assert curSplitSize == 0;
    200.     assert validBlocks.isEmpty();
    201.     assert racks.isEmpty();
    203.     // Process all overflow blocks
    204.     for (OneBlockInfo oneblock : overflowBlocks) {
    205.       validBlocks.add(oneblock);
    206.       curSplitSize += oneblock.length;
    208.       // This might cause an exiting rack location to be re-added,
    209.       // but it should be ok.
    210.       for (int i = 0; i < oneblock.racks.length; i++) {
    211.         racks.add(oneblock.racks[i]);
    212.       }
    214.       // if the accumulated split size exceeds the maximum, then 
    215.       // create this split.
    216.       if (maxSize != 0 && curSplitSize >= maxSize) {
    217.         // create an input split and add it to the splits array
    218.         addCreatedSplit(splits, getHosts(racks), validBlocks);
    219.         curSplitSize = 0;
    220.         validBlocks.clear();
    221.         racks.clear();
    222.       }
    223.     }
    225.     // Process any remaining blocks, if any.
    226.     if (!validBlocks.isEmpty()) {
    227.       addCreatedSplit(splits, getHosts(racks), validBlocks);
    228.     }
    229.   }

    大致过程：

    1. 先遍历每个node，接着遍历每个node的block，在遍历中，将block放入validBlocks数组中，然后将数组中的所有block大小和max_split阈值进行比较，如果大于max_split则将数组中的block取出，合并成一个split。在block遍历结束后对于剩余的block 和min_split大小进行比较，如果大于min_split，则将剩余的block合并成一个split，否则将剩余的block放入nodeToBlocks数组进行缓存，等待rack级别进行合并
    2. 在node遍历结束后，对于同一rack下的剩余block进行合并，合并逻辑通步骤1。

    从上面的分析中可以看出，每个split的大小可能是不一样的。