从sql-client.sh提交的Flink作业如何基于savepoint重启？

Flink的Savepoint仅存储作业运行时的算子状态快照，本身不会持久化提交时的SQL定义、依赖JAR、运行参数等元信息——这是SQL Client提交作业的天然特性，不是异常故障。

前置准备材料 #

重启前必须对齐所有原作业配置，否则无法保证状态映射成功，需要先找回的材料包括：

• 生成Savepoint时对应的完整SQL脚本，包含所有DDL、DML语句、作业运行时的自定义配置项
• 当时SQL Client加载的全部依赖JAR包（包括UDF、非内置的Connector、第三方依赖包）
• 当时启动SQL Client的所有参数：包括-j 指定的JAR路径、-i 指定的初始化文件内容、命令行通过-D 传入的动态配置（并行度、状态后端、Checkpoint配置等）、sql-client-defaults.yaml 里的自定义修改项
  如果SQL脚本遗失，可通过Flink作业历史服务的归档详情页，在原作业的配置列表中找到提交时上传的SQL文本片段拼接；若未开启作业历史服务，可从操作记录、历史命令、业务配置文档中找回，不要凭空重写表结构，极易出现拓扑不匹配问题。

具体重启操作步骤 #

• 第一步：使用和原提交完全一致的参数启动SQL Client
  启动命令对齐原提交的所有配置即可，示例格式：
  ./sql-client.sh -j ./lib/自定义UDF.jar -j ./lib/外置KafkaConnector.jar -i ./init.sql -D state.backend=rocksdb -D execution.parallelism=4
• 第二步：进入SQL交互界面后，先设置Savepoint恢复参数
  执行以下配置，规避SQL Client生成新作业ID、存在未认领临时状态导致的校验失败问题：

  -- 指定Savepoint的绝对路径
  SET execution.savepoint.path = '/opt/flink/savepoints/savepoint-xxxx/xxxxxx';
  -- 允许跳过当前拓扑未认领的临时状态
  SET execution.checkpointing.savepoint.ignore-unclaimed-state = 'true';
  

• 第三步：逐行执行准备好的完整SQL脚本，确保所有DDL、全局配置和原作业完全一致
• 第四步：执行原作业对应的DML语句（如INSERT INTO ... SELECT ...）提交作业，Flink会自动从指定Savepoint加载对应状态，完成作业恢复运行。

关键注意事项 #

• 禁止使用flink run 命令恢复这类作业：SQL Client提交的作业主类是Table框架自动生成的，不存在可手动指定的用户Jar主类，flink run 无法正确构建作业拓扑
• 必须保证本次提交生成的算子拓扑和生成Savepoint时的拓扑完全一致：包括表结构、Connector参数、字段顺序、数据转换逻辑都不能修改，否则会导致状态映射失败
• 如果作业涉及必须的拓扑调整，需要通过table.exec.uid.* 配置为对应算子设置固定的UID后缀，保证状态可以和新拓扑映射，否则会出现状态找不到的问题
• 不需要手动指定原作业的JobID，只要Savepoint路径正确、拓扑匹配，Flink会自动完成状态加载，新作业会自动分配新的JobID

内容的提问来源于stack exchange，提问作者Norman He