code: @Override
      public void filter(Record record, Writer writer)
              throws IOException, InterruptedException {
          Record emitRecord = new Record();
  
          for (int i = 0; i < joinColumn.length; i++) {
              emitRecord.addGrouping("JOIN_KEY" + i, record.getValueString(joinColumn[i]));
          }
  
          emitRecord.addSort(1, Record.SORT_LOWER, 1);
          for (String s : masterLabels) {
              emitRecord.addValue(s, record.getValueString(s));
          }
          writer.write(emitRecord);
      }
docs: {@inheritDoc}