Flink 1.15 implements SQL script to recover data from savepointh

flinksql Interface creation table , And write the data to , Reference resources ：Flink 1.15 Local cluster deployment Standalone Pattern ( Independent cluster mode )_ One day long -- Hyun Bin's blog -CSDN Blog

  After creating the table, execute the following command to automatically create 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;

 

see job Task list ：

./flink-1.15.0/bin/flink list

preservation savepoint

./bin/flink savepoint :jobId [:targetDirectory]  -- jobid and savepoint Keep the file directory

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

from savepoint/checkPoint recovery

RESET execution.savepoint.path;   -- Reset point Save the path

SET execution.savepoint.path = 'file:///usr/local/flink-1.15.0/flink-savepoints/savepoint-c8fa5a-1161f2626294';  -- Set recovery path

 

 

Execute the following command to recover （ Must and create job At the time of the sql Command consistent ）

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;

If there are any errors reported below （ You can use the following command to ignore the error ）：

set 'execution.savepoint.ignore-unclaimed-state' = 'true'; -- Allow skipping savepoint states that cannot be restored

After success, the interface displays ：

Cancel the task ：

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

./flink-1.15.0/bin/flink cancel c8fa5a96073d064b0a717590a7029c0d

Delete savepoint ：

./bin/flink savepoint --dispose Hold point directory $JOB_ID

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

Stop the job gracefully and create a savepoint ：

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

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

flink sql Refer to official documents for commands ：SQL Client | Apache Flink

flink adopt jar The job created by the package is restored , Start the job from the savepoint ：

Refer to official documentation ：Checkpoints | Apache Flink

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