@Override
  public void reduce(Object key, Iterator values, OutputCollector output,
      Reporter reporter) throws IOException {
    if (reducer.getDone()) {
      return;
    }
    if (oc == null) {
      // propagate reporter and output collector to all operators
      oc = output;
      rp = reporter;
      reducer.setReporter(rp);
      MapredContext.get().setReporter(reporter);
    }

    try {
      BytesWritable keyWritable = (BytesWritable) key;
      byte tag = 0;
      if (isTagged) {
        // remove the tag from key coming out of reducer
        // and store it in separate variable.
        int size = keyWritable.getSize() - 1;
        tag = keyWritable.get()[size];
        keyWritable.setSize(size);
      }

      if (!keyWritable.equals(groupKey)) {
        // If a operator wants to do some work at the beginning of a group
        if (groupKey == null) { // the first group
          groupKey = new BytesWritable();
        } else {
          // If a operator wants to do some work at the end of a group
          LOG.trace("End Group");
          reducer.endGroup();
        }

        try {
          keyObject = inputKeySerDe.deserialize(keyWritable);
        } catch (Exception e) {
          throw new HiveException(
              "Hive Runtime Error: Unable to deserialize reduce input key from "
              + Utilities.formatBinaryString(keyWritable.get(), 0,
              keyWritable.getSize()) + " with properties "
              + keyTableDesc.getProperties(), e);
        }

        groupKey.set(keyWritable.get(), 0, keyWritable.getSize());
        LOG.trace("Start Group");
        reducer.startGroup();
        reducer.setGroupKeyObject(keyObject);
      }
      // System.err.print(keyObject.toString());
      while (values.hasNext()) {
        BytesWritable valueWritable = (BytesWritable) values.next();
        // System.err.print(who.getHo().toString());
        try {
          valueObject[tag] = inputValueDeserializer[tag].deserialize(valueWritable);
        } catch (SerDeException e) {
          throw new HiveException(
              "Hive Runtime Error: Unable to deserialize reduce input value (tag="
              + tag
              + ") from "
              + Utilities.formatBinaryString(valueWritable.get(), 0,
              valueWritable.getSize()) + " with properties "
              + valueTableDesc[tag].getProperties(), e);
        }
        row.clear();
        row.add(keyObject);
        row.add(valueObject[tag]);

        try {
          reducer.process(row, tag);
        } catch (Exception e) {
          String rowString = null;
          try {
            rowString = SerDeUtils.getJSONString(row, rowObjectInspector[tag]);
          } catch (Exception e2) {
            rowString = "[Error getting row data with exception " +
                  StringUtils.stringifyException(e2) + " ]";
          }

          // Log the contents of the row that caused exception so that it's available for debugging. But
          // when exposed through an error message it can leak sensitive information, even to the
          // client application.
          LOG.trace("Hive Runtime Error while processing row (tag="
              + tag + ") " + rowString);
          throw new HiveException("Hive Runtime Error while processing row", e);
        }
      }

    } catch (Throwable e) {
      abort = true;
      if (e instanceof OutOfMemoryError) {
        // Don't create a new object if we are already out of memory
        throw (OutOfMemoryError) e;
      } else {
        LOG.error("Reduce failed", e);
        throw new RuntimeException(e);
      }
    }
  }
ExecReducer is the generic Reducer class for Hive. Together with ExecMapper it is
the bridge between the map-reduce framework and the Hive operator pipeline at
execution time. It's main responsibilities are:

- Load and setup the operator pipeline from XML
- Run the pipeline by transforming key, value pairs to records and forwarding them to the operators
- Sending start and end group messages to separate records with same key from one another
- Catch and handle errors during execution of the operators.
ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecReducer.java