如何从检查点恢复Flink Sql作业？

Question

我正在检查Flink Sql Table与kafka连接器是否可以在EXACTLY_ONCE模式下执行，我的方法是创建一个表，设置合理的检查点间隔，并在event_time字段上使用简单的翻滚函数，最后重新启动我的程序。

以下是我的详细进度：

1:创建一个Kafka表

CREATE TABLE IF NOT EXISTS LOG_TABLE(
   id String,
   ...
   ...
   event_timestamp timestamp(3), watermark for event_timestamp as ....
) 

2：启动我的 Flink 作业，如下所示配置

StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        environment.getCheckpointConfig().setCheckpointInterval(30000L);
        environment.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        environment.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        environment.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///tmp/checkpoint/"));
        environment.setStateBackend(new HashMapStateBackend());
        environment.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();

TableEnvironment tableEnvironment = StreamTableEnvironment.create(environment, settings);
tableEnvironment.getConfig().getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        tableEnvironment.getConfig().getConfiguration().setString("table.exec.emit.early-fire.delay", "1s");

3：执行我的sql

select tumble_end(event_timestamp, interval '5' minute), 
       count(1) 
       from LOG_TABLE 
       group by tumble(event_timestamp, interval '5' minute)

如我们所见，翻转窗口间隔为5分钟，检查点间隔为30秒，每个翻转窗口触发6个检查点。

在这种情况下，窗口状态丢失：

1. 下午2:00:00，午餐作业，发送100条消息。（作业id为bd208afa6599864831f008d429a527bb，chk1-3触发成功，检查点dir创建检查点文件）
2. 下午2:01:40，关闭我的工作，修改Checkpoint Storage目录为 /tmp/checkpoint/bd208afa6599864831f008d429a527bb/chk-3
3. 下午2:02:00，重新启动作业并发送另一条100消息。
   

所有消息都在2分钟内发送，因此从检查点重新启动后，作业输出应为200，但结果为100，作业失去了第一个作业的状态。我的进度有什么错误吗？请帮忙查一下，谢谢。

Answer 1

在保留精确一次保证的同时重新启动Flink作业需要以特殊方式启动后续作业，以便新作业从恢复上一个作业的状态开始。（修改检查点存储目录，如您在步骤2中所做的那样，没有帮助。）

如果使用 SQL 客户端启动作业，请参阅从保存点启动 SQL 作业，这涉及执行类似如下操作

SET 'execution.savepoint.path' = '/tmp/flink-savepoints/...';

在启动需要恢复状态的查询之前。

如果您使用的是表API，那么详细信息取决于您如何启动作业，但是您可以使用命令行执行以下操作

$ ./bin/flink run \
      --detached \ 
      --fromSavepoint /tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab \
      ./examples/streaming/StateMachineExample.jar

或者您可能正在使用REST API，在这种情况下，您将向< code>/jars/:jarid/run发送配置了< code > save path 的内容。

请注意，您可以使用保留的检查点而不是保存点来重新启动或重新缩放作业。另请注意，如果更改查询的方式使旧状态与新查询不兼容，则这些都不起作用。有关该主题的更多信息，请参阅 FLIP-190。

Flink Operations Playground 是一个教程，更详细地介绍了此主题和相关主题。