我们直到flink被称为有状态的流处理引擎，所谓状态，就是指flink流处理中所保持的一些状态数据，比方说我们要统计当前用户是否在流中已经出现过，就需要在流中保存一个状态，此状态记录着所有已经出现过的用户。
一般而言，流作业可能会一直运行下去，如果因为某些意外导致任务失败，在恢复任务时，我们期望能从最近的存档点恢复，而不是从流作业开始的地方恢复（试想一下，如果打马里奥的时候，每挂一次，就要从开头重新闯关，这会很让人崩溃）。checkpoint就是flink提供的存档点。我们想想，checkpoint中存放什么才能让flink作业恢复呢？答案就是flink作业中的状态。
实际上，flink还提供了一个存档点，叫savepoint，两者之间的不同之处在于，checkpoint是flink定时帮我们存档，而savepoint是我们手动触发存档。两者产生的存档点都可以帮助恢复作业。
flink checkpoint实例：
public class CheckpointWordCount {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置60秒进行一次checkpoint
env.enableCheckpointing(60000);
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///home/wxwmd/checkpoints");
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
DataStreamSource source = env.socketTextStream("ubuntu", 12345, "n");
DataStream> result = source.map(new TransMapFunction()) // 第一步，将输入的单词转换为二元组
.keyBy(x -> x.f0)
.map(new WordMapFunction());
result.print();
env.execute("checkpoint wordcount");
}
static class TransMapFunction implements MapFunction>{
@Override
public Tuple2 map(String s) throws Exception {
return Tuple2.of(s, 1);
}
}
static class WordMapFunction extends RichMapFunction, Tuple2> {
ValueState count;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
ValueStateDescriptor countDescriptor = new ValueStateDescriptor("count", Integer.class);
count = getRuntimeContext().getState(countDescriptor);
}
@Override
public Tuple2 map(Tuple2 value) throws Exception {
int index = count.value()==null? 0: count.value();
count.update(index+1);
return Tuple2.of(value.f0, index);
}
}
}
在这个程序中，我们源源不断地接收ubuntu主机上12345端口传来的单词，并在状态中记录这是第几个相同的单词。我们设置了checkpoint，放在file:///home/wxwmd/checkpoints文件夹下。
我们将这段代码打包放到flink上运行，打包过程详见flink打包运行详解。
checkpoint运行效果
直接打开端口，起起来任务，运行效果：
看到我们的checkpoint也顺利地被保存在了设置的目录下。这里目录的组成是：
file:/home/wxwmd/checkpoints/7cf7acfb725a17d50e0e90562c737733/chk-3
${设置的目录}/${任务id}/chk-${当前是这个任务的第几个checkpoint}
可以看到，我代码里面设置了60s进行一次checkpoint，现在运行了6分29秒，触发了6个checkpoint，完成了6个checkpoint，当前目录是file:/home/wxwmd/checkpoints/7cf7acfb725a17d50e0e90562c737733/chk-6。
从checkpoint中恢复flink任务
- 我们取消任务：
- 我们使用checkpoint恢复任务
从checkpoint中恢复job的命令是：
./flink run -s file:///home/wxwmd/checkpoints/7cf7acfb725a17d50e0e90562c737733/chk-10/_metadata -c com.ms.chk.CheckpointWordCount /home/wxwmd/softwares/flink/wxw_jobs/checkpoint-1.0-jar-with-dependencies.jar
# 格式是：
flink run -s ${checkpoint路径} -c ${MainClass 入口} ${jar包路径}
起起来之后就可以在flink web ui上看到这个任务了。
ps ：我想通过flink web ui提交任务的时候带上-s ${checkpoint路径}来从checkpoint中恢复任务（毕竟能用web ui谁也不想用命令行），但是发现并不能成功，任务是启动了，但是并没有加载checkpoint中的状态。不知道是不是bug，暂且先使用命令行吧。
- 进行测试
可以看出，我再次在端口12345中输入单词的时候，并没有从0开始计数，说明确实是从checkpoint中进行恢复了。
