code: public SingleOutputStreamOperator<T> reduce(ReduceFunction<T> reducer) {
  		return transform("Keyed Reduce", getType(), new StreamGroupedReduce<T>(
  				clean(reducer), getType().createSerializer(getExecutionConfig())));
  	}
docs: Applies a reduce transformation on the grouped data stream grouped on by
  the given key position. The {@link ReduceFunction} will receive input
  values based on the key value. Only input values with the same key will
  go to the same reducer.
  
  @param reducer
  The {@link ReduceFunction} that will be called for every
  element of the input values with the same key.
  @return The transformed DataStream.