I have a Flink pattern that I think could be useful. I have one input Datastream and key it into a KeyedStream. I'd like to chain process functions on that keyed stream to keep logic compartmentalized better. Is there any way to achieve this behavior with introducing extra KeyBy operators?

I also have been finding this limiting - in order to keyby again, you have to preserve the key through the process function so that you can key again, rather than just being able to consider the values. Something like mapValues() in spark KeyedStreamT, K being and extension of DataStream<Tuple2K, T> processValues, etc. It feels cumbersome to have to keep the key info passed through manually and makes the aggregate functions more complex than they need to be. Several of the examples in documentation show reducers or aggregate functions that would be useless in practice because the key isn’t kept around, like this AverageAggregate which is supposed to get the average over the keyed window, but the resulting datastream is just a DataStream Double without any info of what key the average applies to.

1. Wouldn’t

DataStreamUtils.reinterpretAsKeyedStream()

do what you need? You can insert this after a process function, and you get back to a

KeyedStream

. Of course you better not mess with the keys 🙂 2. When using a reduce or aggregate function, where you need the key in the result, you want to use the version of the reduce/aggregate API that also takes a

ProcessWindowFunction

parameter. This second function gives you access to the result’s key.

I think I might be failing to see the nuance, but how is DataStreamUtils.reinterpretAsKeyedStream() different from just calling keyBy on the DataStream? I think I’m seeing this more as a usability/expressiveness problem. See the example I gave from the docs. The solutions you gave all require you to manually pass the key information through your process or aggregate functions to their output. I understand that the key is accessible from the processwindowfunction, but creating a tuple to pass it through feels less natural or expressive in some cases. This is evident because the folks who wrote the documentation themselves do not do this in some examples

/**
 * The accumulator is used to keep a running sum and a count. The {@code getResult} method
 * computes the average.
 */
private static class AverageAggregate
    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {
  @Override
  public Tuple2<Long, Long> createAccumulator() {
    return new Tuple2<>(0L, 0L);
  }

  @Override
  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
  }

  @Override
  public Double getResult(Tuple2<Long, Long> accumulator) {
    return ((double) accumulator.f0) / accumulator.f1;
  }

  @Override
  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
  }
}

DataStream<Tuple2<String, Long>> input = ...;

input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .aggregate(new AverageAggregate());

It’s much more natural to be able to output a single type and still have the key. If you were to use this example in real life, it would be pretty useless because what you’d get is a bunch of keyless doubles that don’t mean anything without their key.

Re

reinterpretAsKeyedStream

, yes, it doesn’t do a network shuffle (assumes the data is still partitioned the same).

I'm on pyflink so I don't think I have that option to use that utility

I meant pass it through the aggregate/process window, so if I did an additional step later after that aggregate/process window, you have to pass it through, no?

I’m not sure I understand. Your aggregate function doesn’t need to do anything special to “pass through” the current key, since it’s passed to the

ProcessWindowFunction.process()

method. So you can then output the key in the result record, which is what I’m assuming you’re trying to do.

Sorry, I think my intention was not clear - I’m not saying its impossible. I’m saying it would be nice to have a method on KeyedStream that would produce a keyedStream, or produce a data stream with Tuple2<Key, Value> rather than manually having to output the key using a ProcessWindowFunction. Like, from a framework perspective, it would be much less verbose code/more readable to still have the key automatically in the stream. Again, without needing to make use of a specific process window function that passes it along.

OK - yes, there could be a new

WindowedStream.aggregate()

method that takes an additional

KeySelector<K>

function as a parameter, and returns a

KeyedStream<T,K>

by basically doing a call to

reinterpretAsKeyedStream()

for you. And I guess there could be a

WindowedStream.aggregateWithKey()

method that implicitly adds adds a

ProcessWindowFunction

to access the key, and returns a

KeyedStream<Tuple2<K,T>, K>

as the result. Note that an important aspect of Flink’s

KeyedStream

is that it has to have a

KeySelector

function to extract the key from the record type, so unless that’s fixed (as per my second method above) and thus can be easily derived, it’s got to be explicitly provided to the method.