我们如何在 Flink 中对 WindowedStream 进行自定义操作呢？

Question

我想在Flink中对一个WindowedStream进行一些操作，平均来说。但是预定义的运算非常有限，例如求和、最小值、最大值等。

val windowedStream = valueStream
                          .keyBy(0)
                          .timeWindow(Time.minutes(5))
                          .sum(2) //Change this to average?

假设我想求平均值，我该怎么做？

Answer 1

Flink 没有内置函数来计算 WindowStream 上的平均值。您必须为此实现自定义 WindowFunction。

最有效的方法是实现一个 ReduceFunction 来计算您想要平均的值的计数和总和，然后实现一个 WindowFunction 来计算 [=13] 的结果=] 并计算平均值。使用 ReduceFunction 效率更高，因为 Flink 直接将其应用于传入值。因此它会即时聚合值，而不是在 window 中收集它们。这显着减少了 window.

的内存占用

由于 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)
    }
  )

我们如何在 Flink 中对 WindowedStream 进行自定义操作呢？

How do we perform custom operations on WindowedStream in Flink?

apache-flink

flink-streaming