本文介绍了在Kubernetes上持续部署有状态Apache Flink应用的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我想在Kubernetes上运行Apache Flink(1.11.1)流应用程序。其中文件系统状态后端保存到S3。指向S3的检查点工作正常
args:
- "standalone-job"
- "-s"
- "s3://BUCKET_NAME/34619f2862ce3e5fc91d80eae13a434a/chk-4/_metadata"
- "--job-classname"
- "com.abc.def.MY_JOB"
- "--kafka-broker"
- "KAFKA_HOST:9092"
所以我面临的问题是:
- 我必须手动选择以前的状态目录。有没有可能让它变得更好?
- 作业递增chk目录,但它不使用检查点。意味着我在第一次看到事件时抛出一个新事件,并将其存储到
ListState<String>
,每当我通过GitLab部署更新版本的应用程序时,它都会再次引发此事件。 - 定义state.backend到文件系统后,为什么必须在代码中显式启用检查点?
env.enableCheckpointing(Duration.ofSeconds(60).toMillis());
和env.getCheckpointConfig().enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION);
推荐答案
- 您可能会更喜欢Ververica Platform: Community Edition,它将抽象级别提高到您不必在此级别处理细节的程度。它有一个在设计时考虑到CI/CD的API。
- 我不确定我是否理解您的第二点,但是您的作业在恢复过程中会倒带和重新处理一些数据是正常的。Flink并不保证只处理一次,而是保证只处理一次:每个事件将只影响Flink管理的状态一次。这是通过回滚到最近检查点中的偏移量,并将所有其他状态回滚到使用这些偏移量之前的所有数据来实现的。
- 有一个状态后端是必要的,因为它可以在作业运行时存储作业的工作状态。如果不启用检查点,则工作状态将不会被检查点,并且无法恢复。但是,从Flink 1.11开始,您可以使用 通过配置文件启用检查点
execution.checkpointing.interval: 60000
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
这篇关于在Kubernetes上持续部署有状态Apache Flink应用的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!