code: @Override
      public void summarize(Writer writer)
              throws IOException, InterruptedException {
          if (!hasNext()) {
              return;
          }
  
          Record r = next(writer);
          Record masterRecord = new Record();
          for (String s : masterLabels) {
              masterRecord.addValue(s, r.getValueString(s));
          }
  
          if (onlyJoin) {
              while (hasNext()) {
                  Record record = next(writer);
                  Record emitRecord = new Record();
                  for (String s : valueLabels) {
                      emitRecord.addGrouping(s, record.getValueString(s));
                  }
                  for (String s : masterLabels) {
                      emitRecord.addGrouping(s, masterRecord.getValueString(s));
                  }
                  emitRecord.setValueNothing(true);
                  writer.write(emitRecord);
              }
          } else {
              while (hasNext()) {
                  Record record = next(writer);
                  Record emitRecord = new Record();
                  for (int i = 0 ;; i++) {
                      String s = record.getGroupingString("JOIN_KEY" + i);
                      if (s == null) {
                          break;
                      }
                      emitRecord.addGrouping("JOIN_KEY" + i, s);
                  }
                  for (String s : valueLabels) {
                      emitRecord.addValue(s, record.getValueString(s));
                  }
                  for (String s : masterLabels) {
                      emitRecord.addValue(s, masterRecord.getValueString(s));
                  }
                  writer.write(emitRecord);
              }
          }
      }
docs: {@inheritDoc}