public Multimap<Integer, InputSplit> group(Configuration conf,
      Multimap<Integer, InputSplit> bucketSplitMultimap, int availableSlots, float waves,
                                             SplitLocationProvider splitLocationProvider)
      throws IOException {

    // figure out how many tasks we want for each bucket
    Map<Integer, Integer> bucketTaskMap =
        estimateBucketSizes(availableSlots, waves, bucketSplitMultimap.asMap());

    // allocate map bucket id to grouped splits
    Multimap<Integer, InputSplit> bucketGroupedSplitMultimap =
        ArrayListMultimap.<Integer, InputSplit> create();

    // use the tez grouper to combine splits once per bucket
    for (int bucketId : bucketSplitMultimap.keySet()) {
      Collection<InputSplit> inputSplitCollection = bucketSplitMultimap.get(bucketId);
      Class inputFormatClass = conf.getClass("mapred.input.format.class",
              HiveInputFormat.class);
      if (inputFormatClass != BucketizedHiveInputFormat.class &&
              inputFormatClass != HiveInputFormat.class) {
        // As of now only these two formats are supported.
        inputFormatClass = HiveInputFormat.class;
      }
      InputSplit[] rawSplits = inputSplitCollection.toArray(new InputSplit[0]);
      InputSplit[] groupedSplits =
          tezGrouper.getGroupedSplits(conf, rawSplits, bucketTaskMap.get(bucketId),
                  inputFormatClass.getName(), new ColumnarSplitSizeEstimator(), splitLocationProvider);

      LOG.info("Original split count is " + rawSplits.length + " grouped split count is "
          + groupedSplits.length + ", for bucket: " + bucketId);

      for (InputSplit inSplit : groupedSplits) {
        bucketGroupedSplitMultimap.put(bucketId, inSplit);
      }
    }

    return bucketGroupedSplitMultimap;
  }
group splits for each bucket separately - while evenly filling all the
available slots with tasks
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/SplitGrouper.java