code: public <K, V> KVStreamlet<K, V> keyBy(SerializableFunction<R, K> keyExtractor,
                                          SerializableFunction<R, V> valueExtractor) {
      checkNotNull(keyExtractor, "keyExtractor cannot be null");
      checkNotNull(valueExtractor, "valueExtractor cannot be null");
  
      KeyByStreamlet<R, K, V> retval =
          new KeyByStreamlet<R, K, V>(this, keyExtractor, valueExtractor);
      addChild(retval);
      return new KVStreamletShadow<K, V>(retval);
    }
docs: Return a new KVStreamlet<K, V> by applying key and value extractor to each element of this
  Streamlet
  @param keyExtractor The function applied to a tuple of this streamlet to get the key
  @param valueExtractor The function applied to a tuple of this streamlet to extract the value