【详细介绍及演示】Flink之checkpoint检查点的使用
目录
4、查看最近完成的flink job对应的savepoint
一、介绍
Checkpoint:快照点, 是Flink中所有有状态的Operator在某一个时刻的State快照信息/存档信息。
一句话概括: Checkpoint就是State的快照。
二、 设置checkpoint检查点演示
简单的举例说明：
1、 代码演示
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
原文地址：https://blog.csdn.net/m0_57764570/article/details/144092414
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！