是的，我希望从mysql加载初始的状态，因为我的kafka消息是从某个时间点开始的，在这个时间点之前的数据需要先加载到flink state 那现在对于这种场景有什么替代方案吗
> 在 2019年4月16日，下午8:33，Congxian Qiu <[email protected]> 写道： > > Hi > 如果你希望程序在刚开始运行的时候从外部存储加载数据，这个暂时做不到，不过现在社区正在做 Savepoint Reader/Writer > 相关的事情，到时候就可以了 > > Best, Congxian > On Apr 16, 2019, 20:27 +0800, zhang yue <[email protected]>, wrote: >> 你好，我有一个keyed >> state，当我启动flink程序时，我需要根据数据库中的数据给每个key加载不同的数值状态，我应该怎样来操作呢，在文档上找不到例子，我觉得应该是在open函数中设置初始状态，但是在这里获取不到key。 >> >> >> class LineItemStat extends RichFlatMapFunction<ObjectNode, JSONObject> { >> >> /** >> * The ValueState handle. The first field is the count, the second field a >> running sum. >> */ >> private transient MapState<String, Long> stat_value; >> >> @Override >> public void flatMap(ObjectNode input, Collector<JSONObject> out) throws >> Exception { >> >> // access the state value >> >> } >> >> @Override >> public void open(Configuration config) { >> MapStateDescriptor<String, Long> descriptor = >> new MapStateDescriptor<String, Long>( >> "stat_value",String.class, Long.class); // default value of the state, if >> nothing was set >> stat_value = getRuntimeContext().getMapState(descriptor); >> } >> } >>
