本文主要是介绍Flink CheckPoint状态点恢复与savePoint机制，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
Flink CheckPoint状态点恢复与savePoint机制
2019-01-21 22:40:41 大数据技术与架构 阅读数 203更多
分类专栏： Flink从入门到实践
转载自：https://blog.csdn.net/shenshouniu/article/details/84558874
欢迎加入大数据学习群：
**Flink学习视频：**http://edu.51cto.com/sd/88e07
欢迎关注微信号，获取更多关于Spark、flink、hadoop的资料下载。
1 Flink 应用程序启动
./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 -c streaming.SoetWindowWordCountJavaCheckPoint（入口类）
/usr/local/install/testJar/FlinkExample-1.0-SNAPSHOT-jar-with-dependencies.jar（jar路径）
--port 9010
2 Checkpoint 保存与恢复
2.1 Checkpoin设置与保存
- 默认情况下，如果设置了Checkpoint选项，则Flink只保留最近成功生成的1个Checkpoint，而当Flink程序失败时，可以从最近的这个Checkpoint来进行恢复。但是，如果我们希望保留多个Checkpoint，并能够根据实际需要选择其中一个进行恢复，这样会更加灵活，比如，我们发现最近4个小时数据记录处理有问题，希望将整个状态还原到4小时之前
- 1
- 2
- 3
- 4
Flink可以支持保留多个Checkpoint，需要在Flink的配置文件conf/flink-conf.yaml中，添加如下配置，指定最多需要保存Checkpoint的个数。
这样设置以后就查看对应的Checkpoint在HDFS上存储的文件目录
hdfs dfs -ls hdfs://namenode:9000/flink/checkpoints
如果希望回退到某个Checkpoint点，只需要指定对应的某个Checkpoint路径即可实现
2.2 Checkpoint恢复
- 如果Flink程序异常失败，或者最近一段时间内数据处理错误，我们可以将程序从某一个Checkpoint点进行恢复
- -s 后面接的就是待恢复checkpoint的路径。
bin/flink run -s hdfs://namenode:9000/flink/checkpoints/467e17d2cc343e6c56255d222bae3421/chk-56/_metadata flink-job.jar
程序正常运行后，还会按照Checkpoint配置进行运行，继续生成Checkpoint数据
3 SavePoint 剖析
3.1 全局一致性快照
- Flink通过Savepoint功能可以做到程序升级后，继续从升级前的那个点开始执行计算，保证数据不中断
- 全局，一致性快照。可以保存数据源offset，operator操作状态等信息
- 可以从应用在过去任意做了savepoint的时刻开始继续消费
3.2 checkpoint理论
- 应用定时触发，用于保存状态，会过期
- 内部应用失败重启的时候使用
3.3 savePoint 理论
- 用户手动执行，是指向Checkpoint的指针，不会过期，在升级的情况下使用
- 注意：为了能够在作业的不同版本之间以及 Flink 的不同版本之间顺利升级，强烈推荐通过 uid(String) 方法手动的给算子赋予 ID，这些 ID 将用于确定每一个算子的状态范围。如果不手动给各算子指定 ID，则会由 Flink 自动给每个算子生成一个 ID。
- 只要这些 ID 没有改变就能从保存点（savepoint）将程序恢复回来。而这些自动生成的 ID 依赖于程序的结构，并且对代码的更改是很敏感的。因此，强烈建议用户手动的设置 ID。
3.4 savePoint的使用
- 版权声明：本套技术专栏是作者（秦凯新）平时工作的总结和升华，通过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。版权声明：禁止转载，欢迎学习。QQ邮箱地址：1120746959@qq.com，如有任何问题，可随时联系。
- 1：在flink-conf.yaml中配置Savepoint存储位置
不是必须设置，但是设置后，后面创建指定Job的Savepoint时，可以不用在手动执行命令时指定Savepoint的位置：
state.savepoints.dir: hdfs://namenode:9000/flink/savepoints-
- 1
- 1
- 2：触发一个savepoint【直接触发或者在cancel的时候触发】
bin/flink savepoint jobId [targetDirectory] [-yid yarnAppId]【针对on yarn模式需要指定-yid参数】
-
- 1
bin/flink cancel -s [targetDirectory] jobId [-yid yarnAppId]【针对on yarn模式需要指定-yid参数】
这篇关于Flink CheckPoint状态点恢复与savePoint机制的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
