Flink通过savepoint重启作业后ValueState值重置问题如何解决

问题原因 #

1. 命令使用错误 #

这是最核心的原因，两条执行命令都存在参数错误：

• 生成Savepoint的flink stop命令使用错误：--savepointPath参数是用来指定Savepoint的存储目录，你直接将作业ID写在了该参数后，没有传入要停止的作业ID，大概率没有生成合法的Savepoint。
• 恢复作业的flink run命令使用错误：指定从Savepoint恢复的参数是--fromSavepoint（可简写为-s），你使用的--savepointPath不是恢复作业的合法参数，导致作业直接以全新状态启动，没有读取已有的Savepoint。

2. 代码拼写错误 #

如果你的实际运行代码和贴出的内容一致，存在变量名拼写错误：
你定义的状态变量名为persistence，但在flatMap方法最后累加状态的代码写为persistencia.update(persistence.value() + 1);，拼写多了字母a（混用了英语和西语的持久化拼写），会导致状态根本没有成功更新，即使Savepoint生成正确，恢复后也会出现计数异常。

3. 潜在序列化兼容问题 #

你keyBy返回的键类型为Object，如果键的序列化方式在作业重启前后发生变化，会导致相同的业务key匹配不到之前存储的状态，触发状态初始化。

解决方案 #

• 修正执行命令：
  生成Savepoint的正确命令：

  ../bin/flink stop --savepointPath <Savepoint存储目录路径> <作业ID：f74c92af01ed51af94e530ee0e208d7c>
  

  从Savepoint恢复的正确命令：

  ../bin/flink run --fromSavepoint file:/{...}/savepoint-f74c92-6acdb05afd11 flink-andy-12.3.0.jar
  

• 修正代码拼写错误：
  将flatMap方法最后一行的persistencia.update修改为persistence.update，保证状态可以正常累加。

• 优化keyBy逻辑：
  给keyBy指定明确的键类型，避免泛型类型擦除导致的序列化兼容问题。

内容的提问来源于stack exchange，提问作者Andres Jalife