当前位置：首页 > article >正文

flink学习（12）——checkPoint

article 2024/12/3 0:36:50

如何设置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相关的配置
手动保存和自动保存的结果文件是不一样的