我有一系列关于资源的事件，如下所示：

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);
    }
}

从代码中可以明显看出，它将为每个事件发出一个新的resourcetoals，但我希望每分钟为每个资源发出一次total，而不是更频繁。
我试着用一个全局窗口和一个触发器（continuousprocessingtimetrigger）来做实验，但是没有成功。我遇到的问题是：
如何表达我想要的最后一个事件的窗口？
如何不存储在这个全局窗口中生成的所有resourcetoals？
任何帮助都将不胜感激。