我有一系列有关资源的事件,如下所示:

id, type,      count
1,  view,      1
1,  download,  3
2,  view,      1
3,  view,      1
1,  download,  2
3,  view,      1


我正在尝试按每种资源生成统计信息(总计),因此,如果我得到如上所示的流,结果应为:

id, views, downloads
1,  1,     5
2,  1,     0
3,  2,     0


现在,我编写了一个ProcessFunction来计算总数,如下所示:

public class CountTotals extends ProcessFunction<Event, ResourceTotals> {
    private ValueState<ResourceTotals> totalsState;

    @Override
    public void open(Configuration config) throws Exception {
        ValueStateDescriptor<ResourceTotals> totalsDescriptor = new ValueStateDescriptor<>("totals state", ResourceTotals.class);
        totalsDescriptor.setQueryable("resource-totals");
        totalsState = getRuntimeContext().getState(totalsDescriptor);
    }

    @Override
    public void processElement(Event event, Context ctx, Collector<ResourceTotals> out) throws Exception {
        ResourceTotals totals = totalsState.value();
        if (totals == null) {
            totals = new ResourceTotals();
            totals.id = event.id;
        }
        switch (event.type) {
            case "view":
                totals.views += event.count;
                break;
            case "download":
                totals.downloads += event.count;
        }
        totalsState.update(totals);
        out.collect(totals);
    }
}


从代码中可以明显看出,它将为每个事件发出一个新的ResourceTotals,但是我希望每分钟一次而不是更频繁地发出每个资源总计。

我尝试通过使用全局窗口和触发器(ContinuousProcessingTimeTrigger)进行实验,但无法使其正常工作。我遇到的问题是:


如何表达我想要窗口的最后一个事件?
如何最终不存储该全局窗口中曾经产生的所有ResourceTotal?


任何帮助,将不胜感激。

最佳答案

您可以使用计时器每分钟发出一次totalsState中的值。由于我在数据流中看不到任何时间戳,因此我想您会使用一个处理时间计时器。

一种替代方法是用TimeWindow替换ProcessFunction,并使用ReduceFunction保留最后一个事件。

无论哪种情况,您都可以考虑同时使用ID和类型字段来对流进行键控,这应该可以简化状态管理。

更新:

是的,计时器是Flink检查点并恢复状态的一部分。

09-09 22:53
查看更多