在Kubernetes上使用Apache Flink - 如果JobManager崩溃，如何恢复作业

Question

我想在kubernetes上运行一个flink作业，并使用（持久性）状态后端，如果任务管理器崩溃，似乎没有问题，因为它们可以询问作业管理器需要从哪个检查点进行恢复，如果我理解正确的话。

似乎崩溃的作业管理器要更加困难。在flip-6页面上我看到需要zookeeper才能知道作业管理器需要使用哪个检查点进行恢复和进行领导者选举。

由于kubernetes会在作业管理器崩溃时重新启动它，所以是否有一种方法使新的作业管理器无需设置zookeeper集群即可恢复作业呢？

我们目前正在寻找的解决方案是：当kubernetes想要终止作业管理器（例如因为它要将其移动到另一个虚拟机）并创建保存点时，但这只适用于优雅的关闭。

编辑：http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-HA-with-Kubernetes-without-Zookeeper-td15033.html 看起来很有趣，但没有后续跟进。

- Richard Deurwaarder

Answer 1

默认情况下，Flink需要ZooKeeper集群来从JobManager崩溃中恢复。然而，我认为您可以有一个轻量级的实现HighAvailabilityServices、CompletedCheckpointStore、CheckpointIDCounter和SubmittedJobGraphStore，这可以帮助您走得更远。

假设您始终只有一个JobManager在运行（不确定K8s是否能够保证此项要求），并且您有一个持久存储位置，那么您可以实现一个CompletedCheckpointStore，从持久存储系统中检索已完成的检查点（例如读取所有存储的检查点文件）。此外，您将拥有一个包含当前检查点ID计数器CheckpointIDCounter和所有提交的作业图形SubmittedJobGraphStore的文件。因此，基本思路是将所有内容存储在可由单个JobManager访问的持久卷上。

- Till Rohrmann

---

K8s可以保证持久存储位置仅挂载在一个pod / jobmanager上。因此，这也可以保证只有一个jobmanager处于活动状态，这看起来很有前途。如果我实现了这一点，我该如何说服Flink使用这个高可用性服务而不是Zookeeper？ - Richard Deurwaarder

你需要实现自己的 HighAvailabilityServices 并用它来实例化 Flink。以下 PR 可以帮助你：https://github.com/apache/flink/pull/5530。它允许提供自定义的 HighAvailabilityServices 实现。 - Till Rohrmann

你有什么进展吗？我也遇到了同样的情况。谢谢。 - Zack Bartel

@Xeli，你成功地让自定义的HighAvailabilityServices工作了吗？这样在Kubernetes上重新启动Flink集群后，作业就会重新启动了吗？我也想实现同样的功能，如果你能分享一下你的代码，我将不胜感激。 - sceee

@sceee 我已经开始了一个轻量级的实现来推进这个项目，看起来并不太难。但我从未完成它。我有一个新项目即将启动，我想在那时再次拾起它，届时我会更新这个问题 :) - Richard Deurwaarder

@Xeli，你最终完成了吗？ 考虑实现它，但如果已经完成了，我很乐意使用它 :) - ronlut

Answer 2

我基于Till的答案和Xeli的部分实现，实现了一个轻量级的基于文件的HA版本。
您可以在这个github仓库中找到代码 - 在生产中运行良好。
此外，我还写了一系列博客，解释如何在k8s上运行作业集群以及关于这个基于文件的HA实现的具体内容。

- ronlut

Answer 3

对于所有对此感兴趣的人，我目前正在评估和实施一种类似的解决方案，使用Kubernetes ConfigMaps和blob存储（例如S3）来持久化作业元数据，以便在JobManager重启时保留状态。无需使用本地存储，因为该解决方案依赖于持久化到blob存储的状态。

Github thmshmm/flink-k8s-ha

仍有一些工作要做（持久化Checkpoint状态），但基本实现效果非常好。

如果有人想要使用多个JobManagers，Kubernetes提供了一个接口来进行领导者选举，可以利用这个接口。

- Thms Hmm