如何设置checkPoint
package com.bigdata.day06;
/**
* 1、需要三句话
* 2、设置完checkPoint后若程序出现异常，会一直重启
* 3、此时是自动进行checkPoint保存
* 4、注意：此时如果有checkpoint ,是不会出现异常的，需要将checkpoint的代码关闭，再重启程序。
* 5、若配置了
*/
public class _01_checkPointDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 第一句：开启快照，每隔1s保存一次快照
env.enableCheckpointing(1000);
// 第二句：设置快照保存的位置
env.setStateBackend(new FsStateBackend("hdfs://bigdata01:9820/flink/checkpoint"));
// 第三句： 通过webui的cancel按钮，取消flink的job时，不删除HDFS的checkpoint目录
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
DataStreamSource<String> source = env.socketTextStream("bigdata01", 8889);
source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] s = value.split(" ");
for (String string : s) {
out.collect(Tuple2.of(string, 1));
}
}
}).keyBy(0).sum(1).print();
env.execute();
}
}
如何使用checkpoint中记录的数据
运行jar包——使用命令的方式
flink run -c com.bigdata.day06._01_checkPointDemo \
-s hdfs://bigdata01:9820/flink/checkpoint/1ae08d23a3ab0f8455d997c6c913c8e7/chk-95 \
flinkProject-1.0-SNAPSHOT.jar
-s 表示checkPoint的文件地址
最好以下面的方式复制checkPoint的文件地址
设置checkpoint和使用checkpoint是两个事情
手动保存checkPoint
1、使用 ctrl+c是不能停止flink任务的，需要在8081界面进行canle job
2、使用命令手动保存checkPoint 此时保存快照且不会停止任务
flink savepoint 79f53c5c0bb3563b6b6ed3011176c411 hdfs://bigdata01:9820/flink/checkpoint
3、停止的时候，保存一次快照
flink stop --savepointPath hdfs://bigdata01:9820/flink/checkpoint c81bb1546866566ce4e16181d0c52be8
4、停止一个任务
flink stop 152e493da9cdeb327f6cbbad5a7f8e41 job编号
79f53c5c0bb3563b6b6ed3011176c411 job编号
手动保存时，不需要在代码中设置有关checkpoint相关的配置
手动保存和自动保存的结果文件是不一样的
