@Override
  protected void reduce(K row, Iterable<Put> vals, Context context)
    throws IOException, InterruptedException {
    // Using HeapSize to create an upper bound on the memory size of
    // the puts and flush some portion of the content while looping. This
    // flush could result in multiple Puts for a single rowkey. That is
    // acceptable because Combiner is run as an optimization and it's not
    // critical that all Puts are grouped perfectly.
    long threshold =
      context.getConfiguration().getLong("putcombiner.row.threshold", 1L * (1 << 30));
    int cnt = 0;
    long curSize = 0;
    Put combinedPut = null;
    Map<byte[], List<ExtendedCell>> combinedFamilyMap = null;
    for (Put p : vals) {
      cnt++;
      if (combinedPut == null) {
        combinedPut = p;
        combinedFamilyMap = ClientInternalHelper.getExtendedFamilyCellMap(combinedPut);
      } else {
        for (Entry<byte[], List<ExtendedCell>> entry : ClientInternalHelper
          .getExtendedFamilyCellMap(p).entrySet()) {
          List<ExtendedCell> existCells = combinedFamilyMap.get(entry.getKey());
          if (existCells == null) {
            // no cells for this family yet, just put it
            combinedFamilyMap.put(entry.getKey(), entry.getValue());
            // do not forget to calculate the size
            for (ExtendedCell cell : entry.getValue()) {
              curSize += cell.heapSize();
            }
          } else {
            // otherwise just add the cells to the existent list for this family
            for (ExtendedCell cell : entry.getValue()) {
              existCells.add(cell);
              curSize += cell.heapSize();
            }
          }
        }
        if (cnt % 10 == 0) {
          context.setStatus("Combine " + cnt);
        }
        if (curSize > threshold) {
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format("Combined %d Put(s) into %d.", cnt, 1));
          }
          context.write(row, combinedPut);
          combinedPut = null;
          curSize = 0;
          cnt = 0;
        }
      }
    }
    if (combinedPut != null) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(String.format("Combined %d Put(s) into %d.", cnt, 1));
      }
      context.write(row, combinedPut);
    }
  }
Combine Puts. Merges Put instances grouped by <code>K</code> into a single instance.
@see TableMapReduceUtil
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/PutCombiner.java