通过一个小例子学习一下Flink有状态的source、checkpoint和savepoint的使用。代码地址：https://github.com/henshao/flink-learning。例子使用FsStateBackend，savepoint保存在/tmp/flink/checkpoints目录中。。

env.enableCheckpointing(1000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.setStateBackend(new FsStateBackend("file:///tmp/flink/checkpoints"));
env.execute("StatefulSource");

checkpoint

保存

如果在控制台cancel任务，checkpoint都会被清空。所以不要cancel任务，而是停止集群，这样checkpoint目录的数据还会保留着。

$ tree                                                                                                                     
.
├── 21a531af31f6aeddff96f375abd60493
│   ├── chk-14
│   │   ├── 0fa3db43-a58b-47d7-8aa5-67d66f14d96d
│   │   └── _metadata
│   ├── chk-15
│   ├── shared
│   └── taskowned
├── 79362a5fc98b3d62c3d53e4dbae8681f
│   ├── chk-2
│   │   ├── 004e95c6-d381-494b-9e7e-65afacbfe85d
│   │   └── _metadata
│   ├── chk-3
│   ├── shared
│   └── taskowned
├── 7e04c0ae4ab086e497dcdbb81a54a7c1
│   ├── chk-3
│   │   ├── 2e79396f-1162-4f57-a0ad-8e0799031db7
│   │   └── _metadata
│   ├── chk-4
│   ├── shared
│   └── taskowned
└── 99c68672dec699ea9a85231a5ca472fa
    ├── chk-2
    │   ├── _metadata
    │   └── a6d866d8-dec1-4280-98ff-d167423b8f4e
    ├── chk-3
    ├── shared
    └── taskowned

20 directories, 8 files

恢复

$ flink run -s /tmp/flink/checkpoints/7e04c0ae4ab086e497dcdbb81a54a7c1/chk-3/_metadata -c com.henshao.flink.StatefulSource target/flink-learning-1.0-SNAPSHOT.jar

从控制台可以看到任务是从7e04c0ae4ab086e497dcdbb81a54a7c1这个checkpoint恢复的。

image.png

从日志可以看出offset不是0开始的，而是从上次停止的时候开始的。

image.png

savepoint

保存

$ flink savepoint 21a531af31f6aeddff96f375abd60493 /tmp/flink/savepoints/                                  
Triggering savepoint for job 21a531af31f6aeddff96f375abd60493.
Waiting for response...
Savepoint completed. Path: file:/tmp/flink/savepoints/savepoint-21a531-ea92d4b06909
You can resume your program from this savepoint with the run command.

恢复

$ flink run -s /tmp/flink/savepoints/savepoint-21a531-ea92d4b06909 -c com.henshao.flink.StatefulSource target/flink-learning-1.0-SNAPSHOT.jar

从控制台可以看到任务是从savepoint-21a531-ea92d4b06909恢复的。

image.png

参考链接

1. Working with State
2. Flink Checkpoint、Savepoint配置与实践