在 processBroadcastElement 函数中访问 flink 状态 [英] Access flink state inside processBroadcastElement function

问题描述

将在 processBroadcastElement() 函数内部进行一些状态管理.

Going to do some state management inside processBroadcastElement() function.

final val actvTagsMapValue = new MapStateDescriptor[String, List[String]]("actvTagsMapValue", classOf[String], classOf[List[String]])

override def processBroadcastElement(...): Unit {
    val actvTagMap = getRuntimeContext.getMapState(actvTagsMapValue)
    val st = actvTagMap.entries() // this line produce an error
}

在访问状态期间出现以下错误

Getting following error during access state

229797 [LabelShlfEvents -> Sink: Print to Std. Out (1/1)] WARN  
o.a.flink.runtime.taskmanager.Task - LabelShlfEvents -> Sink: Print to Std. Out (1/1) 
(d3154841fd8bd4cabc00e0145ac37ed8) switched from RUNNING to FAILED. 
java.lang.NullPointerException: No key set. This method should not be called outside of a keyed context.
at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75)

我不被允许这样做?

推荐答案

你不能这样做.原因很简单，如 MapState(还有 ValueState、ListState 和更多如 此处) 是一种称为键控状态的状态.此状态被分区并限定为当前元素的输入键.

You can't do this. The reason is quite simple, as MapState (but also ValueState, ListState and more as described here) is type of state called keyed state. This state is partittioned and scoped to the input key of current element.

广播元素没有以任何方式键控或分区，因此没有附加到这些元素的 KeyedContext.当您尝试访问 processBroadcastElement 内部的状态时，Flink 不知道此请求的作用域是哪个键，这就是为什么您会收到异常.

Broadcast elements are not keyed nor partitioned in any way, so there is no KeyedContext attached to those elements. When You try to access the state inside the processBroadcastElement, Flink has no idea which key is this request scoped to, that's why You will get an exception.

另一方面，您可以安全地在 KeyedBroadcastProcessFunction 的 processElement 中使用键控状态，因为这些元素将分配键并且在键控状态的情况下范围是已知的.

On the other hand, You can safely use keyed state in processElement of KeyedBroadcastProcessFunction, because those element will have key assigned and the scope is known in case of keyed state.

如果您需要为非广播状态的广播元素使用状态，您需要将其实现为文档中描述的操作员状态.

If You need to use state for broadcast elements that is not Broadcast state, You need to implement this as operator state as described in the documentation.

这篇关于在 processBroadcastElement 函数中访问 flink 状态的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！