石家庄 网站建站河南公司网站建设
如何设置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>>() {@Overridepublic 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相关的配置
手动保存和自动保存的结果文件是不一样的