最终获得结果的 Flink 会话窗口

Question

我有一个类似于以下模式的 kafka 消息：

{ user: 'someUser', value: 'SomeValue' , timestamp:000000000}

使用 Flink 流计算对这些项目执行一些计数操作。

现在我想声明一个会话，在X秒范围内收集相同的用户+值作为一个单一的，具有最新的时间戳，然后它将被转发到下一个流一次

所以我写了这样的东西：

data.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Data>() {
        .....
    })
    .keyBy(new KeySelector<Data, String>(){

        .......
    })
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .aggregate(new AggregateFunction<Data, Data, Data>() {

        @Override
        public Data createAccumulator() {
            return null;
        }

        @Override
        public Data add(Data value, Data accumulator) {
            if(accumulator == null) {
                accumulator = value;
            }
            return accumulator;

        }

        @Override
        public Data getResult(Data accumulator) {
            return accumulator;
        }

        @Override
        public Data merge(Data a, Data b) {
            return a;
        }
   });

但问题是 getResult 函数是在每个元素上调用的，而不仅仅是在窗口的末尾。

我的问题是如何在窗口结束之前不将聚合结果转发到下一个流。据我所知，当没有更多元素时，即使窗口没有结束，流程流结果也会向前移动

有什么建议吗？

谢谢

Answer 1

Flink 提供了两种不同的方法来评估窗口。在这种情况下，您想使用另一个。

一种方法是逐步评估每个窗口的内容。这就是您使用reduce 和aggregate 所得到的。当元素被分配给窗口时，ReduceFunction 或 AggregateFunction 被调用并且该元素立即对最终结果做出贡献。

替代方法是将process 与ProcessWindowFunction 一起使用。使用这种方法，在窗口完成之前不会评估窗口，此时会调用一次 ProcessWindowFunction 并使用包含分配给窗口的所有元素的 Iterable。这样做的缺点是需要存储所有元素直到窗口被触发，如果ProcessWindowFunction 必须做大量工作来计算可能会暂时中断管道的结果，但需要进行一些计算方式——比如计算不同的元素。

请参阅documentation 了解更多信息。