apache-flink - 如何统计给定时间窗口内Apache Flink处理的记录数

Question

在flink中定义一个时间窗口后如下:

val lines = socket.timeWindowAll(Time.seconds(5))

如何计算 5 秒的特定窗口中的记录数？

Answer 1

执行计数聚合的最有效方法是 ReduceFunction。但是，reduce 有输入和输出类型必须相同的限制。因此，在应用窗口之前，您必须将输入转换为 Int:

val socket: DataStream[(String)] = ???

val cnts: DataStream[Int] = socket
  .map(_ => 1)                    // convert to 1
  .timeWindowAll(Time.seconds(5)) // group into 5 second windows
  .reduce( (x, y) => x + y)       // sum 1s to count