Apache Flink：如何应用多个计数窗口函数？

Question

我有一串带有键的数据，需要计算不同时间段（1分钟、5分钟、1天、1周）内的滚动计数。

是否可以在单个应用程序中计算所有四个窗口计数？

- david Mclaughlin

Answer 1

是的，这是可能的。

如果您正在使用事件时间，您可以通过逐步增加时间间隔级联窗口来实现。所以您可以这样做：

DataStream<String> data = ...
// append a Long 1 to each record to count it.
DataStream<Tuple2<String, Long>> withOnes = data.map(new AppendOne); 

DataStream<Tuple2<String, Long>> 1minCnts = withOnes
  // key by String field
  .keyBy(0) 
  // define time window
  .timeWindow(Time.of(1, MINUTES))
  // sum ones of the Long field
  // in practice you want to use an incrementally aggregating ReduceFunction and 
  // a WindowFunction to extract the start/end timestamp of the window
  .sum(1);

// emit 1-min counts to wherever you need it
1minCnts.addSink(new YourSink());

// compute 5-min counts based on 1-min counts
DataStream<Tuple2<String, Long>> 5minCnts = 1minCnts
  // key by String field
  .keyBy(0)
  // define time window of 5 minutes
  .timeWindow(Time.of(5, MINUTES))
  // sum the 1-minute counts in the Long field
  .sum(1);

// emit 5-min counts to wherever you need it
5minCnts.addSink(new YourSink());

// continue with 1 day window and 1 week window

请注意，这是可能的，因为：

1. Sum是一个可结合的函数（可以通过对局部和进行求和来计算总和）。 2. 滚动窗口对齐并且不重叠。

关于增量聚合ReduceFunction的评论：

通常，您希望在窗口操作的输出中具有窗口的开始和/或结束时间（否则同一键的所有结果看起来都相同）。窗口的开始和结束时间可以从WindowFunction的apply()方法的窗口参数中访问。但是，WindowFunction不会逐步聚合记录，而是收集它们，并在窗口结束时聚合记录。因此，使用ReduceFunction进行增量聚合，使用WindowFunction将窗口的开始和/或结束时间附加到结果上更有效率。文档详细讨论了细节。

如果您想使用处理时间计算此内容，则不能级联窗口，而必须从输入数据流扇出到四个窗口函数。

- Fabian Hueske

---

非常好 - 一旦我编写了assignTimestampsAndWatermarks等代码，我就可以使用EventTime，这一直是计划中的。您能稍微详细解释一下“最佳实践reduceFunction”注释吗？听起来很有趣... - david Mclaughlin

当然，我扩展了我的回答。 - Fabian Hueske