code: private static <T, K> MapOperatorBase<Tuple2<K, T>, T, ?> translateSelectorFunctionReducer(Keys.SelectorFunctionKeys<T, ?> rawKeys,
  			ReduceFunction<T> function, TypeInformation<T> inputType, String name, Operator<T> input, int dop)
  	{
  		@SuppressWarnings("unchecked")
  		final Keys.SelectorFunctionKeys<T, K> keys = (Keys.SelectorFunctionKeys<T, K>) rawKeys;
  		
  		TypeInformation<Tuple2<K, T>> typeInfoWithKey = new TupleTypeInfo<Tuple2<K, T>>(keys.getKeyType(), inputType);
  		
  		KeyExtractingMapper<T, K> extractor = new KeyExtractingMapper<T, K>(keys.getKeyExtractor());
  		
  		PlanUnwrappingReduceOperator<T, K> reducer = new PlanUnwrappingReduceOperator<T, K>(function, keys, name, inputType, typeInfoWithKey);
  		
  		MapOperatorBase<T, Tuple2<K, T>, GenericMap<T, Tuple2<K, T>>> keyExtractingMap = new MapOperatorBase<T, Tuple2<K, T>, GenericMap<T, Tuple2<K, T>>>(extractor, new UnaryOperatorInformation<T, Tuple2<K, T>>(inputType, typeInfoWithKey), "Key Extractor");
  		MapOperatorBase<Tuple2<K, T>, T, GenericMap<Tuple2<K, T>, T>> keyRemovingMap = new MapOperatorBase<Tuple2<K, T>, T, GenericMap<Tuple2<K, T>, T>>(new KeyRemovingMapper<T, K>(), new UnaryOperatorInformation<Tuple2<K, T>, T>(typeInfoWithKey, inputType), "Key Extractor");
  
  		keyExtractingMap.setInput(input);
  		reducer.setInput(keyExtractingMap);
  		keyRemovingMap.setInput(reducer);
  		
  		// set dop
  		keyExtractingMap.setDegreeOfParallelism(input.getDegreeOfParallelism());
  		reducer.setDegreeOfParallelism(dop);
  		keyRemovingMap.setDegreeOfParallelism(dop);
  		
  		return keyRemovingMap;
  	}
docs: --------------------------------------------------------------------------------------------