3.4、Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 Checkpointing

骑猪看日落 2021-10-18 05:50 219阅读 0赞

Flink中的每个函数和操作符都可以是有状态的(有关详细信息，请参阅//TODO使用状态)。有状态函数在各个元素/事件的处理过程中存储数据，使状态成为任何类型的更精细操作的关键构建块。

为了使状态容错，Flink需要对状态进行检查点。检查点允许Flink恢复流中的状态和位置，从而为应用程序提供与无故障执行相同的语义。

数据流容错详细介绍容错机制背后的技术//TODO

### 基础概念 ###

Flink的检查点机制与流和状态的持久存储交互。一般来说，它要求：

1.  一种持久性(或持久性)数据源，可以在一定时间内重放记录。此类源的例子有持久消息队列(例如Apache Kafka、RabbitMQ、Amazon Kinesis、谷歌PubSub)或文件系统(例如HDFS、S3、GFS、NFS、Ceph，……)。
2.  状态的持久存储，通常是分布式文件系统(例如，HDFS、S3、GFS、NFS、Ceph，…)

### 启用和配置Checkpointing ###

默认情况下，检查点是禁用的。要启用检查点，可以在StreamExecutionEnvironment上调用enablecheckpoint (n)，其中n是检查点间隔(以毫秒为单位)，大多数配置的是5分钟：

其他参数：

*  ***exactly-once vs. at-least-once：***您可以选择将模式传递给enablecheckpoint (n)方法，以在两个保证级别之间进行选择。对大多数应用程序来说，一次是最好的。至少一次可能与某些超低延迟(始终只有几毫秒)应用程序相关。
 *  **checkpoint timeout：**在此之后中止正在进行的检查点的时间(如果到那时检查点还没有完成)。
 *  **minimum time between checkpoints：**为了确保流应用程序在检查点之间取得一定的进展，可以定义在检查点之间需要花费多少时间。例如，如果将此值设置为5000，则无论检查点持续时间和检查点间隔如何，下一个检查点将在前一个检查点完成后5秒内启动。注意，这意味着检查点间隔永远不会小于这个参数。通过定义“time between checkpoints”比检查点间隔更容易配置应用程序，因为“time between checkpoints”不受检查点可能比平均时间更长这一事实的影响(例如，如果目标存储系统暂时变慢)。**注意：**这个值还意味着并发检查点的数量为1。
 *  **number of concurrent checkpoints：**默认情况下，当一个检查点还在进行时，系统不会触发另一个检查点。这确保拓扑不会在检查点上花费太多时间，也不会在处理流方面取得进展。可以允许多个重叠的检查站,这很有趣的管道有一定处理延迟(例如,因为函数调用外部服务,需要一些时间来回应),但仍然想做的非常频繁的检查点(100毫秒)处理文档很少失败。**当定义检查点之间的最小时间间隔时，不能使用此选项。**
 *  **externalized checkpoints：**可以定期检查点配置为在外部持久化。外部检查点将其元数据写到持久存储中，并且在作业失败时不会自动清理。如果你的任务失败了，你就会有一个checkpoint来恢复任务。关于外部化检查点的部署说明中有更多细节请查看[Checkpoints][]//TODO。
 *  **fail/continue task on checkpoint errors：**在执行checkpoint的过程中发生错误，判断任务是否失败。这是一个默认的行为。开启之后会忽略chenkpoint失败，继续运行任务。

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);

### 相关的配置选项 ###

可以通过conf/flink-conf设置更多的参数和/或默认值。yaml(参见[配置获得完整的指南][Link 1]//TODO)：

<table> 
 <tbody> 
  <tr> 
   <td style="width:240px;">名称</td> 
   <td style="width:74px;">默认值</td> 
   <td style="width:535px;">描述</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.backend</p> </td> 
   <td style="width:74px;">(none)</td> 
   <td style="width:535px;">用于存储和检查点状态</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.backend.async</p> </td> 
   <td style="width:74px;">true</td> 
   <td style="width:535px;">否应在可能且可配置的情况下使用异步快照方法。一些状态后端可能不支持异步快照，或者只支持异步快照，并忽略此选项。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.backend.fs.memory-threshold</p> </td> 
   <td style="width:74px;">1024</td> 
   <td style="width:535px;">状态数据文件的最小大小。所有小于该值的状态块都内联存储在根检查点元数据文件中。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.backend.incremental</p> </td> 
   <td style="width:74px;">false</td> 
   <td style="width:535px;">是否应该创建增量检查点。对于增量检查点，只存储前一个检查点的差异，而不存储完整的检查点状态。一些状态后端可能不支持增量检查点并忽略此选项。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.backend.local-recovery</p> </td> 
   <td style="width:74px;">false</td> 
   <td style="width:535px;">配置本地恢复。默认情况下，本地恢复被禁用。本地恢复目前只覆盖键控状态后端。目前，MemoryStateBackend不支持本地恢复，因此忽略该选项。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.checkpoints.dir</p> </td> 
   <td style="width:74px;">(none)</td> 
   <td style="width:535px;">用于在Flink支持的文件系统中存储检查点的数据文件和元数据的默认目录。存储路径必须可从所有参与的进程/节点(即所有任务管理器和作业管理器)。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.checkpoints.num-retained</p> </td> 
   <td style="width:74px;">1</td> 
   <td style="width:535px;">保留的已完成检查点的最大数量。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>state.savepoints.dir</p> </td> 
   <td style="width:74px;">(none)</td> 
   <td style="width:535px;">保存点的默认目录。用于将保存点写入文件系统的状态后端(MemoryStateBackend, FsStateBackend, RocksDBStateBackend)。</td> 
  </tr> 
  <tr> 
   <td style="width:240px;"> <p>taskmanager.state.local.root-dirs</p> </td> 
   <td style="width:74px;">(none)</td> 
   <td style="width:535px;">定义根目录的配置参数，用于存储用于本地恢复的基于文件的状态。本地恢复目前只覆盖键控状态后端。目前，memorystateback不支持本地恢复，因此忽略该选项</td> 
  </tr> 
 </tbody> 
</table>

### 选择状态后端 ###

Flink的检查点机制将所有状态的一致快照存储在定时器和有状态操作符中，包括连接器、窗口和任何用户定义的状态。检查点存储在哪里(例如，JobManager memory, file system, database)取决于配置的**状态后端**。

默认情况下，状态保存在任务管理器的内存中，检查点存储在JobManager的内存中。对于大状态的适当持久性，Flink支持在其他状态后端存储和检查点状态的各种方法。状态后端选择可以通过`StreamExecutionEnvironment.setStateBackend(…)`配置。

有关可用状态后端以及作业范围和集群范围配置的选项的详细信息，请参阅[状态后端][Link 2]//TODO。

### 迭代作业中的状态检查点 ###

Flink目前仅为没有迭代的作业提供处理保证。在迭代作业上启用检查点会导致异常。为了在迭代程序上强制检查点，用户需要在启用检查点时设置一个特殊的标志：`env.enableCheckpointing(interval, CheckpointingMode.EXACTLY_ONCE, force = true)`

### 重新启动策略 ###

Flink支持不同的重启策略，这些策略控制在发生故障时如何重启作业。有关更多信息，请参见[重启策略][Link 3]//TODO。

[Checkpoints]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/checkpoints.html#externalized-checkpoints
[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/config.html#full-reference
[Link 2]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html
[Link 3]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/restart_strategies.html