Flink 1.10.1用FsStateBackend崩溃重启后无法加载状态求助

一、核心问题：本地模式需显式指定Checkpoint恢复路径 #

Flink集群模式会自动识别可用的Checkpoint，但本地运行时（非集群）不会自动扫描目录加载Checkpoint，必须在启动时明确告诉应用要从哪个Checkpoint恢复。

修改Checkpoint配置代码 #

在你的环境初始化代码中，添加恢复逻辑，通过系统参数接收Checkpoint路径：

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);

// 新增：从系统参数读取Checkpoint恢复路径，启动时指定
String restoreCheckpointPath = System.getProperty("flink.restore.checkpoint");
if (restoreCheckpointPath != null && !restoreCheckpointPath.isEmpty()) {
    System.out.println("从Checkpoint恢复：" + restoreCheckpointPath);
    env.restoreCheckpoint(restoreCheckpointPath);
}

启动应用时指定恢复路径 #

重启Jar包时，通过-D参数传入你要恢复的Checkpoint目录（比如chk-123）：

java -Dflink.restore.checkpoint=/some/path/checkpoints/chk-123 -jar your-app.jar

二、修复EventCountMap中的错误 #

你的RichMapFunction有两个会影响状态恢复的问题：

1. static的eventCounter会导致并发问题：多个Task实例会共享这个静态变量，状态数据会混乱。
2. 不必要的数据库加载逻辑：状态恢复成功后，previous_state会自动加载Checkpoint中的数据，完全不需要从数据库重新导入。

修改后的EventCountMap代码：

public class EventCountMap extends RichMapFunction<Event, EventCounter> {
    private static final MapStateDescriptor<String, Timestamp> descriptor = new MapStateDescriptor<>("previous_counter", String.class, Timestamp.class);
    private MapState<String, Timestamp> previous_state;
    private static final StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(org.apache.flink.api.common.time.Time.days(1))
            .cleanupFullSnapshot()
            .build();

    // 提前初始化TTL，避免在open方法中重复调用
    static {
        descriptor.enableTimeToLive(ttlConfig);
    }

    @Override
    public void open(Configuration parameters) {
        // 这里获取的MapState会自动从Checkpoint恢复数据
        previous_state = getRuntimeContext().getMapState(descriptor);
    }

    // 删掉从数据库加载状态的mapRefueled方法

    @Override
    public EventCounter map(Event event) throws Exception {
        // 每次map都创建新的EventCounter，避免并发冲突
        EventCounter eventCounter = new EventCounter();
        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;
    }
}

三、验证Checkpoint有效性 #

先确认你的Checkpoint目录/some/path/checkpoints下存在类似chk-xxx的文件夹，里面包含metadata文件——这说明Checkpoint已经成功生成，具备恢复条件。

四、注意事项 #

• 恢复时，作业的拓扑结构（算子数量、状态描述符名称、类型）必须和生成Checkpoint时完全一致，否则无法恢复。
• 如果后续修改了状态的结构（比如MapState的键值类型），旧的Checkpoint将无法复用。

内容的提问来源于stack exchange，提问作者Alter