滑动时间窗口的Flink性能问题

我正在尝试使用一些网络监视器工作。我的目标是不同的计数dst_ip每src_ip。

我的以下代码有效，但性能非常糟糕。似乎每个滑动窗口都会重新计算所有事件，但这不是必需的。

例如，我们有事件按时间秒1 - 600.Flink可以得到每秒的累加器，所以我们每秒有600个累加器。当第一个滑动窗口到期时，flink只合并1-300的累加器，并销毁第二个1的累加器。该窗口也可以在最后一秒之前预合并1-299。当第二个滑动窗口到期时，flink只合并2-301的累加器，并销毁第二个2的累加器。依此类推......

这种方式比将事件分配给多个窗口更有效，并计算每个窗口的聚合。

flink支持吗？我可以用flink自己获得类似的功能吗？

public static class AverageAccumulator2 {

String key;
Set<String> target;
AverageAccumulator2() {
    target = new HashSet<>();
}

}

public static class Average2 implements AggregateFunction>> {

@Override
public AverageAccumulator2 createAccumulator() {
    return new AverageAccumulator2();
}

@Override
public AverageAccumulator2 add(ObjectNode value, AverageAccumulator2 accumulator) {
    accumulator.key = value.get("value").get("src_ip").asText();
    accumulator.target.add(value.get("value").get("dst_ip").asText());
    return accumulator;
}
@Override
public Tuple3<String, Long, Set<String>> getResult(AverageAccumulator2 accumulator) {
    return new Tuple3<>(accumulator.key, (long) accumulator.target.size(), accumulator.target);
}

@Override
public AverageAccumulator2 merge(AverageAccumulator2 a, AverageAccumulator2 b) {
    a.target.addAll(b.target);
    return a;
}

}

final SingleOutputStreamOperator> > process2 =
stream.keyBy(value -> value.get("value").get("sip").asText())

                .timeWindow(Time.seconds(300),Time.seconds(1))
                .aggregate(new Average2());