【详细介绍及演示】Flink之checkpoint检查点的使用-EW帮帮网

package com.bigdata.checkpoint;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class CheckPointWordCountDemo {

    public static void main(String[] args) throws Exception {

        //1. env-准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 在windows运行，将数据提交hdfs,会出现权限问题，使用这个语句解决。
        System.setProperty("HADOOP_USER_NAME", "root");
        // 在这个基础之上，添加快照
        // 第一句：开启快照，每隔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);

        //checkpoint默认重启策略是一直重启,可以自己定义重启策略
        //重启策略可以单独使用，不设置checkpoint也可使用   //savepoint可以手动使用命令设置checkpoint
        //2分钟内重启3次，重启时间间隔是5s
        env.setRestartStrategy(
                RestartStrategies.failureRateRestart(3,
                        Time.of(2, TimeUnit.MINUTES),
                        Time.of(5,TimeUnit.SECONDS))
        );

        //2. source-加载数据
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        //3. transformation-数据处理转换
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return Tuple2.of(word,1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> tuple2) throws Exception {
                return tuple2.f0;
            }
        }).sum(1).print();
        //4. sink-数据输出

        //5. execute-执行
        env.execute();
    }
}

2、测试代码效果

首先启动本地的nc，启动hdfs服务

3、查看快照情况

运行，刷新查看checkpoint保存的数据，它会先生成一个新的文件夹，然后再删除老的文件夹，在某一时刻，会出现两个文件夹同时存在的情况。

三、在集群上运行

首先启动flink：start-cluster.sh

由上一步可以发现数据是保存了，但是并没有起作用，想起作用需要在集群上运行，以下演示集群上的效果：

1、第一次运行

在本地先clean, 再package ，再Wagon一下：

在bigdata01服务器上执行以下命令

#flink run -c 全类名 /opt/app/flink-test-1.0-SNAPSHOT.jar

flink run -c com.bigdata.day06._01CheckPointDemo /opt/app/FlinkDemo-1.0-SNAPSHOT.jar

#记得，先启动nc ,再启动任务，否则报错！

通过nc -lk 9999 输入以下内容：

进入bigdata01:8081页面上查看结果

想查看运行结果，可以通过使用的slot数量判断一下：

取消flink job的运行

查看一下这次的单词统计到哪个数字了：

2、第二次运行

#flink run -c 全类名  -s hdfs://hadoop10:8020/flink-checkpoint/293395ef7e496bda2eddd153a18d5212/chk-34  /opt/app/flink-test-1.0-SNAPSHOT.jar

#启动
flink run -c com.bigdata.day06._01CheckPointDemo -s hdfs://bigdata01:9820/flink/checkpoint/bf416df7225b264fc34f8ff7e3746efe/chk-603  /opt/app/FlinkDemo-1.0-SNAPSHOT.jar

-s 指定从checkpoint目录恢复状态数据，注意每个人都不一样

从上一次离开时，截止的checkpoint目录

观察数据：在nc 上输入一个hello,1 得到新的结果hello,8

四、自定义检查点savePoint

checkpoint自动完成state快照、savePoint是手动的完成快照。

如果程序在没有设置checkpoint的情况，可以通过savePoint设置state快照

1、提交一个flink job 打成jar包

package com.bigdata.checkpoint;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;


public class CheckPointWordCountDemo {

    public static void main(String[] args) throws Exception {

        //1. env-准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //重启策略可以单独使用，不设置checkpoint也可使用  
        //savepoint可以手动使用命令设置checkpoint
        //2分钟内重启3次，重启时间间隔是5s
        env.setRestartStrategy(
                RestartStrategies.failureRateRestart(3,
                        Time.of(2, TimeUnit.MINUTES),
                        Time.of(5,TimeUnit.SECONDS))
        );

        //2. source-加载数据
        DataStreamSource<String> source = env.socketTextStream("bigdata01", 2727);
        //3. transformation-数据处理转换
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] arr = line.split(",");
                for (String word : arr) {
                    collector.collect(word);
                }
            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                //自制一个bug用来测试
                if(word.equals("bug")){
                    throw new Exception("出错了，请重试");
                }
                return Tuple2.of(word,1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> tuple2) throws Exception {
                return tuple2.f0;
            }
        }).sum(1).print();
        //4. sink-数据输出

        //5. execute-执行
        env.execute();
    }
}

执行改任务

2、输入一些数据，观察单词对应的数字的变化

3、执行savepoint操作，添加检查点

停止flink job，并且触发savepoint操作

flink stop --savepointPath  hdfs://bigdata01:9820/flink-savepoint  152e493da9cdeb327f6cbbad5a7f8e41

后面的序号为Job 的ID

不会停止flink的job，只是完成savepoint操作(执行这个操作)

flink savepoint 79f53c5c0bb3563b6b6ed3011176c411 hdfs://bigdata01:9820/flink-savepoint

停止一个 flink 的任务

flink stop 6a27b580aa5c6b57766ae6241d9270ce

后面的序号为Job 的ID

4、查看最近完成的flink job对应的savepoint

发现任务中已经有检查点

5、重新启动flink job，进行测试

停止任务后，查看最终检查点的路径

然后重新启动

flink run -c com.bigdata.day06._01CheckPointDemo -s hdfs://bigdata01:9820/flink-savepoint/savepoint-79f53c-64b5d94771eb 
 /opt/app/flink-test-1.0-SNAPSHOT.jar

-c后是全类名，-s 后是检查的路径，最后一部分是jar包的位置

6、观察变化

再次输入单词，可以看到在之前的基础上累加

另外，在集群中运行我们的程序，默认并行度为1，它不会按照机器的CPU核数，而是按照配置文件中的一个默认值运行的

五、总结

有两种添加检查点的方式：

1、在java代码中自动添加

在执行任务时会在hdfs上创建检查点

// 第一句：开启快照，每隔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);

2、在集群上通过命令在指定位置手动添加

flink savepoint 任务号 hdfs://bigdata01:9820/flink-savepoint

【详细介绍及演示】Flink之checkpoint检查点的使用

一、介绍

二、设置checkpoint检查点演示

1、代码演示

2、测试代码效果

3、查看快照情况

三、在集群上运行

1、第一次运行

2、第二次运行

四、自定义检查点savePoint

1、提交一个flink job 打成jar包

2、输入一些数据，观察单词对应的数字的变化

3、执行savepoint操作，添加检查点

4、查看最近完成的flink job对应的savepoint

5、重新启动flink job，进行测试

6、观察变化

五、总结

网站公告

今日签到

热门文章

最新发布

【详细介绍及演示】Flink之checkpoint检查点的使用

一、介绍

二、 设置checkpoint检查点演示

1、 代码演示

2、测试代码效果

3、查看快照情况

三、在集群上运行

1、第一次运行

2、第二次运行

四、自定义检查点savePoint

1、提交一个flink job 打成jar包

2、输入一些数据，观察单词对应的数字的变化

3、执行savepoint操作，添加检查点

4、查看最近完成的flink job对应的savepoint

5、重新启动flink job，进行测试

6、观察变化

五、总结

网站公告

今日签到

热门文章

最新发布

二、设置checkpoint检查点演示

1、代码演示