Flink 1.7 Google翻译
Flink中的每个函数和运算符都可以是有状态的（有关详细信息，请参阅使用state）。有状态函数在各个元素/事件的处理中存储数据，使状态成为任何类型的更复杂操作的关键构建块。
为了使状态容错，Flink需要检查状态。检查点允许Flink恢复流中的状态和位置，从而为应用程序提供与无故障执行相同的语义。
关于流容错的文档详细描述了Flink的流容错机制背后的技术
先决条件
Flink检查点机制与流和状态的持久化交互，通常，它需要
- 持久化数据源会在特定时间段内重播记录。此类源的示例是持久性消息队列（例如，Apache Kafka，RabbitMQ，Amazon Kinesis，Google PubSub）或文件系统（比如HDFS, S3, GFS, NFS, Ceph, …）
- 状态的持久化存储，特别是分布式文件系统（例如HDFS, S3, GFS, NFS, Ceph, …）
启用和配置检查点
默认的，检查点是关闭的。需要在StreamExecutionEnvironment调用enableCheckpointing(n)开启，n为检查点的时间间隔毫秒数
其他的检查点参数如下：
- exactly-once vs. at-least-once: 您可以选择将模式传递给enableCheckpointing（n）方法，以在两个保证级别之间进行选择。exactly-once是和大多数应用，at-least-once相对适合超低延迟（通常是毫秒级）的应用。
- checkpoint timeout: 检查点超时时间
- minimum time between checkpoints: 为了确保流应用程序在检查点之间取得一定进展，可以定义检查点之间需要经过多长时间。如果将此值设置为例如5000，则无论检查点持续时间和检查点间隔如何，下一个检查点将在上一个检查点完成后的5秒内启动。请注意，这意味着检查点间隔永远不会小于此参数。
通过定义“检查点之间的时间”而不是检查点间隔来配置应用程序通常更容易，因为“检查点之间的时间”不易受检查点有时需要比平均时间更长的事实的影响（例如，如果目标存储系统暂时很慢）。
请注意，此值还表示并发检查点的数量为1。 - number of concurrent checkpoints: 默认情况下，当一个检查点仍在进行时，系统不会触发另一个检查点。这可确保拓扑不会在检查点上花费太多时间，也不会在处理流方面取得进展。可以允许多个重叠检查点，这对于具有特定处理延迟的管道（例如，因为函数调用需要一些时间来响应的外部服务）而感兴趣，但是仍然希望执行非常频繁的检查点（100毫秒） ）在失败时重新处理很少。
当定义检查点之间的最短时间时，不能使用此选项 - externalized checkpoints: 你可以配置定期检查点以在外部持久化。外部化检查点将其元数据写入持久存储，并且在作业失败时不会自动清除。这样，如果你的工作失败，你将有一个检查点来恢复。有关外部化检查点的部署说明中有更多详细信息。
- fail/continue task on checkpoint errors: 这将确定在执行任务的检查点过程时是否发生错误，任务是否会失败。这是默认行为。或者，当禁用此选项时，任务将简单地拒绝检查点协调器的检查点并继续运行
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// start a checkpoint every 1000 ms
env.enableCheckpointing(1000);
// advanced options:
// set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// make sure 500 ms of progress happen between checkpoints
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig().setCheckpointTimeout(60000);
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// enable externalized checkpoints which are retained after job cancellation
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
| 属性 | 默认值 | 描述 |
|---|---|---|
| state.backend | (none) | 存储检查点状态的状态后端 |
| state.backend.async | true | 选择状态后端是否应在可能和可配置的情况下使用异步快照方法。某些状态后端可能不支持异步快照，或仅支持异步快照，并忽略此选项 |
| state.backend.fs.memory-threshold | 2014 | 状态数据文件的最小大小。小于该值的所有状态块都内联存储在根检查点元数据文件中。 |
| state.backend.incremental | false | 如果可能，选择状态后端是否应创建增量检查点。对于增量检查点，仅存储来自先前检查点的差异，而不是完整的检查点状态。某些状态后端可能不支持增量检查点并忽略此选项 |
| state.backend.local-recovery | false | 此选项配置此状态后端的本地恢复。默认情况下，禁用本地恢复。本地恢复目前仅涵盖关键状态后端。目前，MemoryStateBackend不支持本地恢复并忽略此选项 |
| state.checkpoints.dir | (none) | 用于在Flink支持的文件系统中存储检查点的数据文件和元数据的默认目录。必须可以从所有参与的进程/节点（即所有TaskManagers和JobManagers）访问存储路径。 |
| state.checkpoints.num-retained | 1 | 要保留的已完成检查点的最大数量 |
| state.savepoints.dir | (none) | 保存点的默认目录。由将后端写入文件系统的状态后端使用（MemoryStateBackend，FsStateBackend，RocksDBStateBackend） |
| taskmanager.state.local.root-dirs | (none) | config参数定义根目录，用于存储基于文件的状态以进行本地恢复。本地恢复目前仅涵盖关键状态后端。目前，MemoryStateBackend不支持本地恢复并忽略此选项 |
选择状态后端
Flink的检查点机制存储定时器和有状态运算符中所有状态的一致快照，包括连接器，窗口和任何用户定义的状态。存储检查点的位置（例如，JobManager内存，文件系统，数据库）取决于配置的状态后端。
默认情况下，状态保存在TaskManagers的内存中，检查点存储在JobManager的内存中。为了适当持久化大状态，Flink支持在其他状态后端中存储和检查点状态的各种方法。可以通过StreamExecutionEnvironment.setStateBackend（...）配置状态后端的选择。
有关可用状态后端的详细信息以及作业范围和群集范围配置的选项，请参阅状态后端
可迭代Jobs的检查点状态
Flink目前仅为没有迭代的作业提供处理保证。在迭代作业上启用检查点会导致异常。为了强制对迭代程序进行检查点，用户在启用检查点时需要设置一个特殊标志：env.enableCheckpointing（interval，force = true）。
请注意，在失败期间，循环边缘中的记录（以及与它们相关的状态变化）将丢失。
重启策略
Flink支持不同的重启策略，可以控制在发生故障时如何重新启动作业。有关更多信息，请参阅重新启动策略
