code: @Override
  	public void prepare() throws Exception {
  		TaskConfig config = this.taskContext.getTaskConfig();
  		if (config.getDriverStrategy() != DriverStrategy.SORTED_GROUP_REDUCE) {
  			throw new Exception("Unrecognized driver strategy for GroupReduce driver: " + config.getDriverStrategy().name());
  		}
  		final Counter numRecordsIn = this.taskContext.getMetricGroup().getIOMetricGroup().getNumRecordsInCounter();
  		
  		this.serializer = this.taskContext.<IT>getInputSerializer(0).getSerializer();
  		this.comparator = this.taskContext.getDriverComparator(0);
  		this.input = new CountingMutableObjectIterator<>(this.taskContext.<IT>getInput(0), numRecordsIn);
  
  		ExecutionConfig executionConfig = taskContext.getExecutionConfig();
  		this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
  
  		if (LOG.isDebugEnabled()) {
  			LOG.debug("GroupReduceDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
  		}
  	}
docs: --------------------------------------------------------------------------------------------