code: private static <IN, OUT, K> PlanUnwrappingReduceGroupOperator<IN, OUT, K> translateSelectorFunctionReducer(
  			Keys.SelectorFunctionKeys<IN, ?> rawKeys, GroupReduceFunction<IN, OUT> function,
  			TypeInformation<IN> inputType, TypeInformation<OUT> outputType, String name, Operator<IN> input,
  			boolean combinable)
  	{
  		@SuppressWarnings("unchecked")
  		final Keys.SelectorFunctionKeys<IN, K> keys = (Keys.SelectorFunctionKeys<IN, K>) rawKeys;
  		
  		TypeInformation<Tuple2<K, IN>> typeInfoWithKey = new TupleTypeInfo<Tuple2<K, IN>>(keys.getKeyType(), inputType);
  		
  		KeyExtractingMapper<IN, K> extractor = new KeyExtractingMapper<IN, K>(keys.getKeyExtractor());
  		
  		PlanUnwrappingReduceGroupOperator<IN, OUT, K> reducer = new PlanUnwrappingReduceGroupOperator<IN, OUT, K>(function, keys, name, outputType, typeInfoWithKey, combinable);
  		
  		MapOperatorBase<IN, Tuple2<K, IN>, GenericMap<IN, Tuple2<K, IN>>> mapper = new MapOperatorBase<IN, Tuple2<K, IN>, GenericMap<IN, Tuple2<K, IN>>>(extractor, new UnaryOperatorInformation<IN, Tuple2<K, IN>>(inputType, typeInfoWithKey), "Key Extractor");
  
  		reducer.setInput(mapper);
  		mapper.setInput(input);
  		
  		// set the mapper's parallelism to the input parallelism to make sure it is chained
  		mapper.setDegreeOfParallelism(input.getDegreeOfParallelism());
  		
  		return reducer;
  	}
docs: --------------------------------------------------------------------------------------------