Flink初入门-容错处理

向右看齐 2021-12-22 18:53 227阅读 0赞

**目录**

检查点：

启用和配置检查点：

相关的配置选项：

选择状态后台：

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

重启策略：

--------------------

## 检查点： ##

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

\--Flink中每个函数和运算符都是有状态的(有关详细信息，请[参阅state][state])。有状态函数在各个元素/事件处理中存储数据，使状态成为任何类型的复杂操作的关键构件块。

为了使状态容错，Flink需要检查状态。将检查点允许Flink恢复流中的状态和位置，从而为应用程序提供无故障执行相同的语义。挂关于流容错的文档详细描述了Flink的流容错机制背后的技术。

\------先决条件

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

> 一个持久(或耐用，可以重放记录在一定量的事件)的数据源。这些源的示例是持久消息队列.(例如，Kafka，RabbitMQ。。。)或文件系统(例如，HDFS,S3,GF3,NFS....)
> 
> 状态的持久存储，通常是分布式文件系统（例如，HDFS,S3,GF3,NFS）

## 启用和配置检查点： ##

默认情况下，禁用检查点。为了使检查点，调用enableCheckpointing(n)上StreamExecutionEnvironment，其中N是以毫秒为单位的检查点间隔。

检查点的其它参数包括：

> 1.完全一次与至少一次；您可以选择将模式传递给enableCheckpointing（n）方法，已在两个保证级别之间进行选择。对于大多数应用来说，恰好一次是优选的。至少一次可能与某些超低延迟(始终未几毫秒)的应用程序相关。
> 
> 2.checkpoint timeout(检查点超时)：如果当前检查点未完成，则终止检查点的时间。
> 
> 3.检查点之间的最短时间：为确保流应用程序在检查点之间取得一定的进展，可以定义检查点之间需要经过多长时间。如果将此值设置为例如5000，则无论检查点持续时间和检查点间隔如何，下一个检查点将在上一个检查点完成后的5秒内启动。请注意，这意味着检查点间隔永远不会小于此参数。通过定义“检查点之间的时间”而不是检查点间隔来配置应用程序通常更容易，因为“检查点之间的时间”不易受检查单有时需要比平均时间更长的事实的影响（“例如，如果在目标存储系统暂时很慢”）。请注意，此值还表示并发检查点的数量为一。
> 
> 4.并发检查点：默认情况下，当一个检查点仍处于运行状态时，系统不会触发另一个检查点。这可确保拓扑不会再检查点上花费过多的时间，也不会在处理流方面取得进展。可以允许多个重叠检查点，这对于具有特定处理延迟的管道（例如，因为函数调用需要一段时间来相应的外部服务）而感兴趣，但是仍然希望执行非常频繁的检查点（100毫秒）在失败时重新处理很好。
> 
> 当定义检查点之间的最短时间时，不能使用此选项。
> 
> 5.外部化检查点：您可以将外围检查点配置为外部持久化。外部化检查点将其元数据写入持久存储，并且在作业失败时不会自定清除。这样，如果您的工作失败，您将有一个检查点可以从中recovery。[有关外部化检查点的部署说明中][Link 1]有更多的详细信息。
> 
> 6.关于检查点错误的fail/continue：这确定如果在执行任务的检查点过程中发生错误，任务是否将失败。这是默认行为。或者，当禁用此选项时，任务将简单地拒绝检查点并继续运行。

scala

val env = StreamExecutionEnvironment.getExecutionEnvironment()
    
    // start a checkpoint every 1000ms
    env.enableCheckpointing(1000)
    
    // advanced options:
    
    // set mode to exactly-once (this is the default)
    env.getCheckpointConfig.setChecckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    
    // make sure 500ms of progress happen between checkpoints
    env.getCheckpointConfig.setMinPausBetweenCheckpoints(500)
    
    // checkpoints have to complete within one minute,or are discarded
    env.getCheckpointConfig.setCheckpointTimeOut(60000)
    
    // prevent the tasks from failing if an error happens in their checkpointing,the checkpoint will just be declined
    env.getCheckpointConfig.setFailTaskOnCheckpointingError(false)
    
    // allow only one checkpoint to be in progress at the same time
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

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

<table style="width:500px;">  相关配置项/默认值conf/flink-conf.yaml  
 <tbody> 
  <tr> 
   <td>key</td> 
   <td>默认</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>state.backend</td> 
   <td>(none)</td> 
   <td>状态后台用于存储和检查点状态</td> 
  </tr> 
  <tr> 
   <td>state.backend.async</td> 
   <td>true</td> 
   <td>选择状态后台是否应在可能和可配置的情况下使用异步SNAPSHOT方法。某些状态后台可能不支持异步SNAPSHOT,或者仅支持异步SNAPSHOT,并忽略此选项。</td> 
  </tr> 
  <tr> 
   <td>state.backend.fs.memory-threshold</td> 
   <td>1024</td> 
   <td>状态数据文件的最小大小。由于改制的所有状态都内联存储在根检查点元数据文件中。</td> 
  </tr> 
  <tr> 
   <td>state.backend.incremental</td> 
   <td>false</td> 
   <td>如果可能，选择状态后台是否应创建增量检查点。对于增量检查点，仅存储来自先前检查点的差异，而不是完整的检查点状态，某些状态后台可能不支持增量检查并忽略此选项。</td> 
  </tr> 
  <tr> 
   <td>state.backend.local-recovery</td> 
   <td>false</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>state.checkpoints.dir</td> 
   <td>(none)</td> 
   <td>用于在Flink支持的文件系统中存储检查点的数据文件和元数据的默认目录。必须可以从所有参与的进程/节点（即所有TM和JobManagers）访问存储路径。</td> 
  </tr> 
  <tr> 
   <td>state.checkpoint.num-retained</td> 
   <td>1</td> 
   <td>要retained的已完成检查点的最大数量</td> 
  </tr> 
  <tr> 
   <td>state.savepoint.dir</td> 
   <td>(none)</td> 
   <td>保存点的默认目录。由将后台写入文件系统的状态后台（MemoryStateBackend，FsStateBackend，RocksDBStateBackendd）使用</td> 
  </tr> 
  <tr> 
   <td>taskmanager.state.local.root-dirs</td> 
   <td>(none)</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

## 选择状态后台：    ##

Flink的检查点机制存储定时器和有状态算子中是所有的一致的SNAPSHOT,包括连接器，窗口和任何用户定义的状态。存储检查点的位置(例如，JobManager内存，文件系统，数据库)取决于配置的**状态后台**。

默认情况下，状态保存在TM的内存中，检查点存储在JobManager的内存中。为了适当持久化大状态，Flink支持在其它状态后台中存储和检查点状态的各种方法。可以通过配置状态后台的选择StreamExecutionEnvironment.setSatteBackend(...)

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

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

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

请注意，在失败期间，循环边缘中的记录（以及与它们相关的状态变化）将丢失。

## 重启策略： ##

Flink支持不同的重启策略，循环边缘中的记录（以及与他们相关的状态变化）将丢失。

[state]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/state/state.html
[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/checkpoints.html#externalized-checkpoints
[Link 2]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html