Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/4626
I think it would be nice to have a utility here in order to make this
easier to use:
```java
DataStream result =
Utils.withPreaggregation(
stream.timeWindow(Time.minutes(5)),
myAggregateFunction
)
.apply(windowFunction);
```
The utility would basically take the aggregate function and insert the
stream transformation for the pre-aggregation on the "*predecessor* or the
keyed stream, and then set up the `WindowedStream` again.
Pseudo code:
```java
public static <T, K, W extends Window, A> WindowedStream<T, K, W>
preaggregate(
WindowedStream<T, K, W> windowedStream,
AggregateFunction<T, A, T> preAggregator) {
// sanity check that the windowedStream has no custom trigger and evictor
PreAggregationOperator preAggOp = new
PreAggregationOperator(preAggregator, properties from windowed stream);
DataStream<T> originalStream = 'get predecessor before keyBy from
windowed stream'
DataStream<T> preAggregated = originalStream.transform(preAggOp , ...);
WindowedStream<T, K, W> windowedAgain = preAggregated
.keyBy(key extractor from original windowed stream)
.window(assigner);
return windowedAgain;
}
```
---
