如何在 Flink 中构建 1 小时的重播流缓冲区

Question

我想动态保留最近 1 小时事件的缓冲区。该缓冲区应该为我提供一个重放函数，以便可以对最后一小时的数据执行查询。Flink 中已经实现了某些东西吗？还是我需要自己构建它？

我尝试使用 Window API，但似乎 Flink 没有给我一个前移固定宽度的时间窗口。

Answer 1

我得到了我自己的问题的解决方案，但我想保留这个问题，以防你有更好的解决方案。因为我的肯定违反了函数式编程的一些良好实践。

我的黑客如下。

            val keyedEventStream: KeyedStream[E]
            // create a stream of [hourly window as a set of events]
            val eventWindowStream = keyedEventStream.timeWindow(Time.minutes(60), Time.milliseconds(50)).fold(scala.collection.Set[E]())((set: scala.collection.Set[E], event: E) => set + event)

            // This is the hourly buffer my process logic will use
            var workWindow = scala.collection.Set[E]()
            // update the workspace window with the stream of hourly window.
            eventWindowStream.map((set: scala.collection.Set[W]) => workWindow = set)

如您所见，最后一个映射的唯一目的是更新变量 workWindow，这实际上是内联函数的副作用......