Flink中最大时间窗口的更新流

Question

source
  .keyBy(...)
  .timeWindow(...)
  .fold((DummyKey, 0)) { case ((_, current), key) => (key, current + 1) }
  .keyBy(_ => ())
  .maxBy(1)

fold的结果是一个(key，count)元素流-因此，我希望从这个流中获得一个“具有最高计数的key”的更新流。

然后我通过一个常量(keyby(_=>())-因为这是一个全局操作）进行键控，并使用maxby-这几乎可以实现：我得到一个最高计数流，但当前的最高计数是针对每个元素发出的。

我想我要找的是某种带有前一个值的过滤器，它只会在新值与前一个值不同时才会发出元素。

目前在Flink有可能吗？

Answer 1

Flink默认情况下没有这样的过滤器，但自己实现一个应该相当容易。

您可以使用类似于以下的有状态flatmap来完成此操作：

val source: DataStream[Int] = ???

source
  .keyBy(_: Int => _)
  .timeWindow(Time.minutes(10))
  .fold((1, 0)) { case ((_, current), key) => (key, current + 1) }
  // move everything to the same key
  .keyBy(_ => 0) 
  // use stateful flatmap to remember highest count and filter by that
  .flatMapWithState( (in, state: Option[Int]) => 
    // filter condition
    if (in._2 > state.getOrElse(-1)) 
      // emit new value and update max count
      (Seq(in), Some(in._2)) 
    else 
      // emit nothing (empty Seq()) and keep count
      (Seq(), state)
  ).setParallelism(1)

如果非并行（单线程）筛选器运算符成为瓶颈，您可以通过添加具有随机键的KeyBy和具有更高并行度的有状态筛选器FlatMap来添加并行预筛选器。