Flink 1.15实现 Sql 脚本从savepointh恢复数据

flinksql界面创建表，并把数据写入，参考：Flink 1.15本地集群部署Standalone模式(独立集群模式)_天一道长--玄彬的博客-CSDN博客

 创建表结束后执行以下命令会自动创建flink job：

INSERT INTO `all_users_sink`(id,user_name,address,phone_number,email,ct_date)

SELECT id,user_name,address,phone_number,email,ct_date FROM user_source;

 

查看job任务列表：

./flink-1.15.0/bin/flink list

保存savepoint

./bin/flink savepoint :jobId [:targetDirectory]  -- jobid 和savepoint保持文件目录

./bin/flink savepoint c8fa5a96073d064b0a717590a7029c0d file:///usr/local/flink-1.15.0/flink-savepoints

从savepoint/checkPoint恢复

RESET execution.savepoint.path;   -- 重置point保存路径

SET execution.savepoint.path = 'file:///usr/local/flink-1.15.0/flink-savepoints/savepoint-c8fa5a-1161f2626294';  -- 设置恢复路径

 

 

在执行以下命令就可以恢复（必须和创建job时的sql命令一致）

INSERT INTO `all_users_sink`(id,user_name,address,phone_number,email,ct_date)

SELECT id,user_name,address,phone_number,email,ct_date FROM user_source;

如果有以下报错（可以用以下命令忽略报错）：

set 'execution.savepoint.ignore-unclaimed-state' = 'true'; --允许跳过无法还原的保存点状态

成功之后的界面显示：

取消任务：

./flink-1.15.0/bin/flink cancel $JOB_ID

./flink-1.15.0/bin/flink cancel c8fa5a96073d064b0a717590a7029c0d

删除保存点：

./bin/flink savepoint --dispose 保持点目录 $JOB_ID

./bin/flink savepoint --dispose /usr/local/flink-1.15.0/flink-savepoints/savepoint-c8fa5a-1161f2626294 c8fa5a96073d064b0a717590a7029c0d

优雅的停止作业并创建保存点：

./bin/flink stop --savepointPath /tmp/flink-savepoints $JOB_ID

./bin/flink stop --savepointPath /usr/local/flink-1.15.0/flink-savepoints 74e0eddfa18eb39c403617f1d573cfdd

flink sql命令参考官方文档：SQL Client | Apache Flink

flink 通过jar包创建的作业恢复，从保存点启动作业：

参考官方文档：Checkpoints | Apache Flink

./bin/flink run --detached --fromSavepoint /usr/local/flink-1.15.0/flink-savepoints/savepoint-74e0ed-e3e4e431bd5d ./examples/streaming/StateMachineExample.jar