为Yarn Cluster中的Flink Job设置Flink Checkpoint的路径动态

Question

我正在使用Yarn来运行Flink作业。对于每个Flink工作，我正在创建一个检查点。

我提交了一个在我的Yarn集群中运行的Flink作业。我有一个轮询工作，检查作业是否在Yarn上失败并重新启动它。再次提交作业时，Yarn会为此Flink作业创建一个新的application_id。如何配置重新提交的Flink作业以使用重新启动的Flink作业的检查点。

我在flink-conf.yaml`中设置了conf state.savepoints.dir = hdfs://localhost:9000/checkpoint/

在创建Flink作业时，streamExecutionEnvironment.setStateBackend(new FsStateBackend("hdfs://localhost:9000/checkpoint/uuid-job-1"));

当我进行此设置时，检查点将保存在conf文件（hdfs://localhost:9000/checkpoint/）中指定的路径中，而不是保存在创建Flink作业时设置的路径中。

任何帮助将不胜感激。谢谢！

Answer 1

不幸的是，你不能用旧的检查站开始新的工作。你能做的就是使用externalized checkpoints。 flink <= 1.5的一个缺点是外部化检查点的元数据存储在由config参数设置的所有作业的单个目录中：state.checkpoints.dir。但您可以在每次提交前更改它。

来自邮件列表thread的另一个注释：

好消息是，Flink 1.5将重新设计外部化检查点的工作方式：基本上，所有检查点现在都可以被视为外部化，元数据将存储在检查点的根目录中，而不是存储在所有作业的一个全局目录中。这样，外化检查点的元数据驻留在每个作业的检查点目录中，从中恢复它应该相当简单。