code: public <R> GroupReduceOperator<T, R> reduceGroup(GroupReduceFunction<T, R> reducer) {
  		if (reducer == null) {
  			throw new NullPointerException("GroupReduce function must not be null.");
  		}
  
  		String callLocation = Utils.getCallLocationName();
  		TypeInformation<R> resultType = TypeExtractor.getGroupReduceReturnTypes(reducer, getType(), callLocation, true);
  		return new GroupReduceOperator<>(this, resultType, clean(reducer), callLocation);
  	}
docs: Applies a GroupReduce transformation on a non-grouped {@link DataSet}.
  
  <p>The transformation calls a {@link org.apache.flink.api.common.functions.RichGroupReduceFunction} once with the full DataSet.
  The GroupReduceFunction can iterate over all elements of the DataSet and emit any
  number of output elements including none.
  
  @param reducer The GroupReduceFunction that is applied on the DataSet.
  @return A GroupReduceOperator that represents the reduced DataSet.
  
  @see org.apache.flink.api.common.functions.RichGroupReduceFunction
  @see org.apache.flink.api.java.operators.GroupReduceOperator
  @see DataSet