背景

Flink状态后端持久化通常有3种方案：内存、文件系统和rocksDb。经常用的是文件系统，一般是分布式文件系统：HDFS。但是测试的时候用分布式文件系统太麻烦了，因为有时只是开发过程中简单测试，希望直接用本地文件系统。

实现

首先得要做checkpoint才能恢复，所以在checkpoint的时候就得持久化到本地文件系统磁盘上：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// ...
// 使用哈希集合状态后端
env.setStateBackend(new HashMapStateBackend());
// 指定持久化的时候路径（本地文件系统）
env.getCheckpointConfig().setCheckpointStorage("file:///E:\\checkpoints");

恢复的时候在 execution.savepoint.path 中指定我们在本地状态存储路径即可。参考示例：

Configuration envConf = new Configuration();
// 替换为自己的checkpoint地址
envConf.setString("execution.savepoint.path", "file:///E:\\checkpoints\\d041cf06aafc7869e93bd70c1f11edc3\\chk-1");
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(envConf);

优化

以上这么写不够优雅，可以结合 ParameterTool ，通过变量的形式在启动的入参指定即可。

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

@Slf4j
public class SavepointDemo {

    public static void main(String[] args) {

        // 使用ParameterTool接收来自外部的变量
        final ParameterTool params = ParameterTool.fromArgs(args);

        // 环境配置
        Configuration envConf = new Configuration();
        // -------------------------------------------------------
        // ----------------  状态恢复  ---------------------------
        // -------------------------------------------------------
        // 获取需要恢复的状态路径
        // 示例：--savepoint.path "file:///E:\\checkpoints\\d041cf06aafc7869e93bd70c1f11edc3\\chk-1"
        String savepointPath = params.get("savepoint.path");
        if (StringUtils.isNotEmpty(savepointPath)) {
            // 打印日志
            log.info("从指定的路径恢复状态：{}", savepointPath);
            //
            envConf.setString("execution.savepoint.path", savepointPath);
        }

        // 获取执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(envConf);

        // -------------------------------------------------------
        // ----------------  状态后端  ---------------------------
        // -------------------------------------------------------
        // 启用 checkpoint
        env.getCheckpointConfig().setCheckpointInterval(30 * 1000);
        // checkpoint持久化目录
        String checkpointFs = params.get("checkpoint.fs");
        if(StringUtils.isNotEmpty(checkpointFs)) {
            // 使用哈希集合状态后端
            env.setStateBackend(new HashMapStateBackend());
            // 指定持久化的时候路径（本地文件系统）
            env.getCheckpointConfig().setCheckpointStorage(checkpointFs);
        }
        
        // 具体作业逻辑 ...

        try {
            env.execute("Savepoint Demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

总结

记录一下，主要经常忘记Window的路径怎么写，盘符和转义字符经常懒得记。