我使用带有FsStateBackend的Flink 1.10.1作为检查点的状态后端。我有一些有状态的操作，在应用程序运行期间(作为.jar应用程序运行，而不是作为集群运行(，它们按预期工作，但如果应用程序由于某种原因停止(或崩溃(，应该存储在带有检查点的文件系统中的状态将不会加载，函数也不会有任何以前的引用，然后我需要从数据库加载信息，并将其保存为状态，以便再次使用这些以前的状态。必须有一种方法可以使用检查点和FsStateBackend来做到这一点，而不必从数据库中读取所有信息，只需从已经存储的检查点中重新加载这些状态即可。这可能吗？
以下是一些代码：我的检查点配置
final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(8, GetConfiguration.getConfig());
final StateBackend stateBackend = new FsStateBackend(new Path("/some/path/checkpoints").toUri(), true);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.enableCheckpointing(30000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);
env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
env.setRestartStrategy(RestartStrategies.noRestart());
env.setStateBackend(stateBackend);
这就是我想避免的例子：
public class EventCountMap extends RichMapFunction<Event, EventCounter> {
private static final MapStateDescriptor<String, Timestamp> descriptor = new MapStateDescriptor<>("previous_counter", String.class, Timestamp.class);
private static final EventCounter eventCounter = new EventCounter();
private MapState<String, Timestamp> previous_state;
private static final StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(org.apache.flink.api.common.time.Time.days(1))
.cleanupFullSnapshot()
.build();
@Override
public void open(Configuration parameters) {
descriptor.enableTimeToLive(ttlConfig);
previous_state = getRuntimeContext().getMapState(descriptor);
}
/*I want to avoid to call this function that load all events from db and pass them to the state to be used. This happens only once but there must be a efficient way to do this in flink.*/
private void mapRefueled() throws Exception {
Preconditions.checkNotNull(previous_state);
for (Map.Entry<String, Timestamp> map : StreamingJob.update_beh_count_ts.entrySet())
previous_state.put(map.getKey(), map.getValue());
StreamingJob.update_beh_count_ts.clear();
}
@Override
public EventCounter map(Event event) throws Exception {
/*Refuel map state in case of failures*/
if (!StreamingJob.update_beh_count_ts.isEmpty()) mapRefueled();
eventCounter.date = new Date(event.timestamp.getTime());
final String key_first = eventCounter.date.toString().concat("_ts_first");
final String key_last = eventCounter.date.toString().concat("_ts_last");
if (previous_state.contains(key_first) && previous_state.contains(key_last)) {
final Timestamp first = (previous_state.get(key_first).after(event.timestamp)) ? event.timestamp : previous_state.get(key_first);
final Timestamp last = (previous_state.get(key_last).before(event.timestamp)) ? event.timestamp : previous_state.get(key_last);
previous_state.put(key_first, first);
previous_state.put(key_last, last);
} else {
previous_state.put(key_first, event.timestamp);
previous_state.put(key_last, event.timestamp);
}
eventCounter.first_event = previous_state.get(key_first);
eventCounter.last_event = previous_state.get(key_last);
return eventCounter;
}
}
希望我能为你解释自己，让你明白我需要做什么。谨致问候！提前谢谢。
要在重新启动作业时从检查点加载状态，必须明确安排，否则作业将在不加载检查点的情况下重新运行。
有关详细信息，请参阅从保存点恢复和从保留的检查点恢复，但其要点是：
bin/flink run -s :checkpointPath [:runArgs]
我的猜测是这还没有完成。
关于如何配置Flink集群进行自动恢复的最佳实践，这取决于您使用的内容(Yarn、Mesos、Kubernetes…(。