Flink如何计算键控窗口的聚合输出

Question

在Flink中，是否可以计算键控窗口的聚合输出？

我们有一个数据流，我们调用byKey（）指定一个由字符和数字组成的字段（例如A01、A02…A10、B01、B02…B10等），就像棋盘上的方块一样。在byKey（）之后，我们调用窗口（TumblingEventTimeWindow.of（Time.days（7）），因此我们创建了一个每周窗口。在此之后，我们调用reduce（），结果我们得到了SingleOutputStreamOperator

现在，我们要对SingleOutputStreamOP进行分组

这是我的代码，您可以看到：

>

我们使用基于Tuple2的keyBy（）

代码窗口（timeIntervalConstructor.newInstance（））基本上创建了一个每周窗口。

我们调用duce（），因此对于每个键，我们都有一个聚合值。

现在我们使用另一个keyBy（），这次键基本上是根据代码A01，...， A10的数量计算的：如果它大于5，我们有一个海类型，如果它小于或等于我们有另一个。

同样，窗口（timeIntervalConstructor.newInstance（））为第二个每周窗口。

最后，在aggregate（）中，我们计算每组的top3。

        .keyBy(new KeySelector<Query2IntermediateOutcome, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> getKey(Query2IntermediateOutcome intermediateOutcome) throws Exception {
                return new Tuple2<String, Integer>(intermediateOutcome.getCellId(), intermediateOutcome.getHourInDate());
            }
        })
        .window(timeIntervalConstructor.newInstance())
        .reduce(new ReduceFunction<Query2IntermediateOutcome>() {
            @Override
            public Query2IntermediateOutcome reduce(Query2IntermediateOutcome t1, Query2IntermediateOutcome t2) throws Exception {
                t1.setAttendance(t1.getAttendance()+t2.getAttendance());
                return t1;
            }
        })
        .keyBy(new KeySelector<Query2IntermediateOutcome, String>() {
            @Override
            public String getKey(Query2IntermediateOutcome query2IntermediateOutcome) throws Exception {
                return query2IntermediateOutcome.getSeaType().toString();
            }
        })
        .window(timeIntervalConstructor.newInstance())
        .aggregate(new Query2FinalAggregator(), new Query2Window())

这个解决方案可以工作，但我真的不喜欢它，因为第二个窗口在前一次触发时接收所有数据，但它每周都会发生，所以第二个窗口一起接收所有数据，并且必须立即运行聚合（）。




Answer 1

我认为将所有这些业务逻辑分解为一个KeyedProcessFunction是相当简单的。然后您可以避免在周末爆发的活动。

以Flink文档中的本教程为例，了解如何用KeyedProcessFunction替换带关键帧的窗口。