具有大状态的Apache Flink中的保存点

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class SavepointExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 定义数据源
        env.addSource(new SourceFunction<Integer>() {
            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                int count = 0;
                while (isRunning) {
                    ctx.collect(count++);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        }).print();

        // 触发保存点
        env.triggerSavepoint("savepoint_path");

        // 恢复保存点
        env.setStateBackend(new FsStateBackend("file:///path/to/savepoint"));
        env.setRestartStrategy(RestartStrategies.noRestart());
        env.execute("Savepoint Example");
    }
}