code: public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key) {
  		Preconditions.checkNotNull(key);
  		return new KeyedStream<>(this, clean(key));
  	}
docs: It creates a new {@link KeyedStream} that uses the provided key for partitioning
  its operator states.
  
  @param key
  The KeySelector to be used for extracting the key for partitioning
  @return The {@link DataStream} with partitioned state (i.e. KeyedStream)