code: public void reduce(Text key, Iterator<Text> values,
                       OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
      String keyStr = key.toString();
      int pos = keyStr.indexOf(ValueAggregatorDescriptor.TYPE_SEPARATOR);
      String type = keyStr.substring(0, pos);
      ValueAggregator aggregator = ValueAggregatorBaseDescriptor
        .generateValueAggregator(type);
      while (values.hasNext()) {
        aggregator.addNextValue(values.next());
      }
      Iterator outputs = aggregator.getCombinerOutput().iterator();
  
      while (outputs.hasNext()) {
        Object v = outputs.next();
        if (v instanceof Text) {
          output.collect(key, (Text)v);
        } else {
          output.collect(key, new Text(v.toString()));
        }
      }
    }
docs: Combines values for a given key.
  @param key the key is expected to be a Text object, whose prefix indicates
  the type of aggregation to aggregate the values.
  @param values the values to combine
  @param output to collect combined values