初始化地图状态的内容

Question

我已经实现了一个具有以下结构的 FlinkRichFunction：

public class MyFunction extends KeyedBroadcastProcessFunction <String, InputType, BroadcastedStateType, OutputType> {
private MapState<String, MyState> myState;              
@Override
public void open(Configuration conf)throws Exception{
myState = getRuntimeContext().getMapState(new MapStateDescriptor<>("state", Types.STRING, Types.POJO(BroadcastedStateType.class)));
}
@Override
public void processElement(InputType value, ReadOnlyContext ctx, Collector<OutputType> out) throws Exception {
MyState state = myState.get(value.ID());
// Do things
}
@Override
public void processBroadcastElement(BroadcastedStateType value, Context ctx, Collector<OutputType> out) throws Exception {
state.put(value.ID(), value.state());   // Update the mapState with value from broadcast
}
// retrieve all the state values and put them in the MapState
private void initialState() throws Exception{
Map<String, MyState> initialValues = ...;
this.cameras.putAll(initialValues);
}
}

mapState变量存储通过BroadcastedStream更新的多个状态。更新在processBroadcastElement()函数中完成。

在作业开始时，我想使用initialState()函数初始化mapState。

问题是我无法在open()函数中使用它(请参阅此处原因(

在这种情况下初始化mapState的正确方法是什么？(在所有情况下，都有 RichFunctions(

Answer 1

你想实现 org.apache.flink.streaming.api.checkpoint.CheckpointedFunction

执行此操作时，将实现两种方法：

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
// called when it's time to save state
myState.clear();
// Update myState with current application state 
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
// called when things start up, possibly recovering from an error
descriptor = new MapStateDescriptor<>("state", Types.STRING, Types.POJO(BroadcastedStateType.class));
myState = context.getKeyedStateStore().getMapState(descriptor);
if (context.isRestored()) {
// restore application state from myState  
}       
}

你在 initializeState(( 方法而不是 open(( 中初始化你的 myState 变量。

Answer 2

我不相信你实际上可以在 initializeState(( 中初始化广播状态。修改广播状态的唯一方法是通过您在 processBroadcastElement 方法中获得的读/写上下文。

但是你可以做的是在initializeState中使用context.isRerestore((来确定KeyedBroadcastProcessFunction是否是第一次初始化，并设置一个瞬态局部变量来记录此信息。然后，第一次调用 processBroadcastElement 方法时，您可以使用此信息来决定在广播状态下存储什么。但是您必须在广播流上发送一些内容才能启动此操作。