flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamUtils.java
/**
	 * Reinterprets the given {@link DataStream} as a {@link KeyedStream}, which extracts keys with the given
	 * {@link KeySelector}.
	 *
	 * <p>IMPORTANT: For every partition of the base stream, the keys of events in the base stream must be
	 * partitioned exactly in the same way as if it was created through a {@link DataStream#keyBy(KeySelector)}.
	 *
	 * @param stream      The data stream to reinterpret. For every partition, this stream must be partitioned exactly
	 *                    in the same way as if it was created through a {@link DataStream#keyBy(KeySelector)}.
	 * @param keySelector Function that defines how keys are extracted from the data stream.
	 * @param typeInfo    Explicit type information about the key type.
	 * @param <T>         Type of events in the data stream.
	 * @param <K>         Type of the extracted keys.
	 * @return The reinterpretation of the {@link DataStream} as a {@link KeyedStream}.
	 */
Reinterprets the given  DataStream as a  KeyedStream, which extracts keys with the given
KeySelector.

IMPORTANT: For every partition of the base stream, the keys of events in the base stream must be
partitioned exactly in the same way as if it was created through a  DataStream#keyBy(KeySelector).

@param stream      The data stream to reinterpret. For every partition, this stream must be partitioned exactly
in the same way as if it was created through a  DataStream#keyBy(KeySelector).
@param keySelector Function that defines how keys are extracted from the data stream.
@param typeInfo    Explicit type information about the key type.
@param          Type of events in the data stream.
@param          Type of the extracted keys.
@return The reinterpretation of the  DataStream as a  KeyedStream.
Reinterprets the given DataStream as a KeyedStream which extracts keys with the given KeySelector . IMPORTANT : For every partition of the base stream the keys of events in the base stream must be partitioned exactly in the same way as if it was created through a DataStream#keyBy ( KeySelector ) . @param stream The data stream to reinterpret . For every partition this stream must be partitioned exactly in the same way as if it was created through a DataStream#keyBy ( KeySelector ) . @param keySelector Function that defines how keys are extracted from the data stream . @param typeInfo Explicit type information about the key type . @param Type of events in the data stream . @param Type of the extracted keys . @return The reinterpretation of the DataStream as a KeyedStream .
public static <T, K> KeyedStream<T, K> reinterpretAsKeyedStream(
		DataStream<T> stream,
		KeySelector<T, K> keySelector,
		TypeInformation<K> typeInfo) {
		PartitionTransformation<T> partitionTransformation = new PartitionTransformation<>(
			stream.getTransformation(),
			new ForwardPartitioner<>());
		return new KeyedStream<>(
			stream,
			partitionTransformation,
			keySelector,
			typeInfo);
	}
public static < T , K > KeyedStream < T , K > reinterpretAsKeyedStream ( DataStream < T > stream , KeySelector < T , K > keySelector , TypeInformation < K > typeInfo ) { PartitionTransformation < T > partitionTransformation = new PartitionTransformation < > ( stream . getTransformation ( ) , new ForwardPartitioner < > ( ) ) ; return new KeyedStream < > ( stream , partitionTransformation , keySelector , typeInfo ) ; }
Reinterprets the given {@link DataStream} as a {@link KeyedStream}, which extracts keys with the given
 {@link KeySelector}.
Reinterprets the given { @link DataStream } as a { @link KeyedStream } which extracts keys with the given { @link KeySelector } .