状态与容错

Checkpoints

参考 查看如何在 Flink 程序中开启和配置 checkpoint。

要了解 checkpoints 和 savepoints 之间的区别,请参阅 。

启用 Checkpoint 后,managed State 将被持久化,来确保故障后 Flink job 恢复的一致性。 Checkpoint 期间 State 存储的位置取决于所选的 Checkpoint 存储。

Flink 开箱即用地提供了两种 Checkpoint 存储类型:

JobManagerCheckpointStorageFileSystemCheckpointStorage

JobManagerCheckpointStorage 将 Checkpoint 快照存储在 JobManager 的堆内存中。

JobManagerCheckpointStorage 的限制:

默认情况下,每个 State 的大小限制为 5 MB。 可以在 的构造函数中修改大小。无论配置的最大 State 大小如何,状态都不能大于 Pekko 框架的大小(请参阅 )。聚合后总的状态大小必须小于 JobManager 的内存上限。

鼓励在以下场景使用 JobManagerCheckpointStorage:

使用很少状态的作业,例如仅包含每次仅存储一条记录(Map、FlatMap、Filter…)的作业。 Kafka 消费者需要很少的 State。

FileSystemCheckpointStorage 配置中包含文件系统 URL(类型、地址、路径), 例如 “hdfs://namenode:40010/flink/checkpoints” 或 “file:///data/flink/checkpoints”。

Checkpoint 时, Flink 会将 State 快照写到配置的文件系统和目录的文件中。 最少的元数据存储在 JobManager 的内存中(或者,高可用性模式下存储在 Checkpoint 的元数据中)。

如果指定了 Checkpoint 目录,FileSystemCheckpointStorage 将用于保存 Checkpoint 快照。

鼓励使用 FileSystemCheckpointStorage 的场景:

所有高可用的场景。

CheckpointConfig config = env.getCheckpointConfig();config.setExternalizedCheckpointRetention(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION); 配置项定义了当作业取消时,对作业 checkpoint 的操作:

ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION:当作业取消时,保留作业的 checkpoint。注意,这种情况下,需要手动清除该作业保留的 checkpoint。ExternalizedCheckpointRetention.DELETE_ON_CANCELLATION:当作业取消时,删除作业的 checkpoint。仅当作业失败时,作业的 checkpoint 才会被保留。

与 相似,checkpoint 由元数据文件、数据文件(与 state backend 相关)组成。可通过配置文件中 “execution.checkpointing.dir” 配置项来指定元数据文件和数据文件的存储路径,另外也可以在代码中针对单个作业特别指定该配置项。

当前的 checkpoint 目录结构(由 FLINK-8531 引入)如下所示:

其中 SHARED 目录保存了可能被多个 checkpoint 引用的文件,TASKOWNED 保存了不会被 JobManager 删除的文件,EXCLUSIVE 则保存那些仅被单个 checkpoint 引用的文件。

注意: Checkpoint 目录不是公共 API 的一部分,因此可能在未来的 Release 中进行改变。

通过配置文件全局配置execution.checkpointing.dir: hdfs:///checkpoints/创建 state backend 对单个作业进行配置与 savepoint 一样,作业可以从 checkpoint 的元数据文件恢复运行(savepoint恢复指南)。注意,如果元数据文件中信息不充分,那么 jobmanager 就需要使用相关的数据文件来恢复作业(参考)。

Copyright © 2088 《一炮特攻》新版本全球首发站 All Rights Reserved.
友情链接