Flink 没有内置函数来计算WindowStream
. 您必须为此实施自定义WindowFunction
。
最有效的方法是实现一个ReduceFunction
计算您想要平均的值的计数和总和,然后实现一个WindowFunction
获取结果ReduceFunction
并计算平均值的后续方法。使用 aReduceFunction
更有效,因为 Flink 直接将它应用于传入的值。因此,它即时聚合值,而不是在窗口中收集它们。这显着减少了窗口的内存占用。
由于 a 的输出ReduceFunction
与其输入具有相同的类型,因此您需要在应用ReduceFunction
.
像下面这样的东西应该可以解决问题:
val valueStream: DataStream[(String, Double)] = ???
val r: DataStream[(String, Double)] = valueStream
// append a 1L for counting
.map(x => (x._1, x._2, 1l))
// key and window stream
.keyBy(0).timeWindow(Time.minutes(5))
.apply(
// ReduceFunction (compute sum and count)
(x: (String, Double, Long), y: (String, Double, Long)) =>
(x._1, x._2 + y._2, x._3 + y._3),
// WindowFunction
(key, window: TimeWindow, input: Iterable[(String, Double, Long)], out: Collector[(String, Double)]) => {
// get first (and only) value
val x: (String, Double, Long) = input.toIterator.next
// compute average as sum / count
out.collect(x._1, x._2 / x._3)
}
)