我已经实现了具有以下结构的Flink RichFunction：
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()函数中完成的。在工作开始时，我想使用
mapState函数初始化initialState()。问题是我不能在
open()函数中使用它（请参阅here原因）在这种情况下，初始化
mapState的正确方法是什么？ （并且在所有情况下都使用RichFunctions） 最佳答案
您要实现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变量。