Apache Flink-timerService，onTimer获取整个事件

Question

我需要一些关于Apache Flink中timerService的帮助。我的用例非常简单，但我没有找到任何明确的答案。

我的程序从源（在我的例子中是rabbitMQ）接收json格式的事件（映射到MyEvent，这里简化了MyEvent）。事件可以立即处理（1），也可以存储以供以后处理（2）。（2） 我认为TimerService是适当的解决方案。在onTimer方法中，我需要整个对象（MyEvent），而不仅仅是键。因此，首先我认为使用整个json作为键，这很好，但不知怎么的，这感觉是错误的，因为在任何示例中，键都不是这样使用的。第二种方法如下：；使用ValueState。但我的密钥不是唯一的，ValueState是每个密钥。同样，我可以用一个对象作为键。。。在onTimer中，我只能访问ctx中事件的密钥。currentKey，而不是事件本身。。。

因此，我的问题是：无论对象是什么样子，我如何存储整个事件以供后续处理？

这是代码（kotlin）

data class MyEvent(val event: String, val secs: Int)

class CountWithTimeoutFunction : KeyedProcessFunction<String, MyEvent, MyEvent>() {

    private lateinit var state: ValueState<MyEvent>

    override fun open(parameters: Configuration?) {
        state = runtimeContext.getState(ValueStateDescriptor("myState", MyEvent::class.java))
    }

    override fun processElement(myEvent: MyEvent, context: Context, collector: Collector<MyEvent>) {
        println("" + Date() + "-processElement-" + myEvent)
        state.update(myEvent)
        context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + myEvent.secs * 1000)
    }

    override fun onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector<MyEvent>) {
        val myEvent = state.value()
        println("" + Date() + "-onTimer-" + ctx.currentKey + " state.value()_" + myEvent)
    }

}

谢谢你的建议。

Answer 1

对于这样的情况，一种常见的、简单的技术是通过向您使用随机数填充的事件添加一个字段来为每个事件提供一个唯一的键。（请注意，执行keyBy（random.nextLong（））是行不通的，因为Flink依赖于键的确定性。）

有时使用的另一种技术是使用MapState，其中键是计时器时间戳，值是等待该计时器的事件列表。当事件到达时，将它们附加到列表中作为时间戳。当计时器启动时，处理列表中的所有内容，然后丢弃它。

第二种方法将使用较少的计时器，但由于处理这些列表的开销，其效率可能较低（至少对于RocksDB state后端而言）。