1、Flink Managing Execution - Task Failure Recovery（Flink Task 故障恢复）

Myth丶恋晨 2023-06-29 02:52 14阅读 0赞

当任务失败时，Flink 需要重新启动失败的任务和其他受影响的任务，将作业恢复到正常状态。

重新启动策略和故障转移策略用于控制任务重新启动。重新启动策略决定是否以及何时可以重新启动失败/受影响的任务。故障转移策略决定应该重新启动哪些任务以恢复作业。

## 重启策略 ##

可以使用默认的重启策略来启动集群，在没有定义特定于作业的重启策略时，总是使用默认的重启策略。如果提交的作业带有重启策略，该策略将覆盖集群的默认设置。

默认的重启策略是通过 Flink 的配置文件 flink-conf.yaml 设置的。配置参数重新启动策略定义了采取哪种策略。如果没有启用检查点，则使用“no restart”策略。如果检查点被激活，并且没有配置重启策略，则使用 `Integer.MAX_VALUE `的固定延迟策略重启尝试。请参阅下面的可用重启策略列表，以了解支持哪些值。

每个重启策略都有自己的一组参数来控制其行为。这些值也在配置文件中设置。每个重新启动策略的描述包含有关各自配置值的更多信息。

<table> 
 <thead> 
  <tr> 
   <th>Restart Strategy</th> 
   <th>Value for restart-strategy</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Fixed delay</td> 
   <td>fixed-delay</td> 
  </tr> 
  <tr> 
   <td>Failure rate</td> 
   <td>failure-rate</td> 
  </tr> 
  <tr> 
   <td>No restart</td> 
   <td>none</td> 
  </tr> 
 </tbody> 
</table>

除了定义默认的重启策略外，还可以为每个 Flink 作业定义一个特定的重启策略。此重新启动策略是通过调用 ExecutionEnvironment 上的 setRestartStrategy 方法以编程方式设置的。注意，这也适用于 StreamExecutionEnvironment。

下面的示例演示如何为作业设置固定延迟重新启动策略。在失败的情况下，系统尝试重新启动作业3次，并在连续的重新启动尝试之间等待10秒。

val env = ExecutionEnvironment.getExecutionEnvironment()
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
      3, // number of restart attempts
      Time.of(10, TimeUnit.SECONDS) // delay
    ))

下面几节描述特定于重启策略的配置选项。

### Fixed Delay Restart Strategy ###

固定延迟重新启动策略尝试给定次数来重新启动作业。如果超过了最大尝试次数，作业最终会失败。在两次连续的重启尝试之间，重启策略等待固定的时间。

通过在flink- confi.yaml中设置以下配置参数，可以将此策略作为默认启用。

restart-strategy: fixed-delay

<table> 
 <thead> 
  <tr> 
   <th style="width:286px;">Configuration Parameter</th> 
   <th style="width:358px;">Description</th> 
   <th style="width:205px;">Default Value</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="width:286px;"><code>restart-strategy.fixed-delay.attempts</code></td> 
   <td style="width:358px;">在作业被声明为失败之前，Flink重试执行的次数。</td> 
   <td style="width:205px;">如果启用检查点，more之为1, 或&nbsp;<code>Integer.MAX_VALUE</code>&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:286px;"><code>restart-strategy.fixed-delay.delay</code></td> 
   <td style="width:358px;">延迟重试意味着在执行失败后，重新执行不会立即开始，而是在一定的延迟之后才开始。当程序与外部系统交互时(例如连接或挂起的事务在尝试重新执行之前应该达到超时)，延迟重试可能很有帮助。</td> 
   <td style="width:205px;"><code>如果启用检查点，默认值为akka.ask.timeout</code>, or 10s&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

For example：

restart-strategy.fixed-delay.attempts: 3
    restart-strategy.fixed-delay.delay: 10 s

固定延迟重启策略也可程序设置：

### Failure Rate Restart Strategy ###

故障率重新启动策略在失败后重新启动作业，但是当超过故障率(每个时间间隔的失败)时，作业最终失败。在两次连续的重启尝试之间，重启策略等待固定的时间。

通过在flink- confi.yaml中设置以下配置参数，可以将此策略作为默认启用。

restart-strategy: failure-rate

<table> 
 <thead> 
  <tr> 
   <th>Configuration Parameter</th> 
   <th>Description</th> 
   <th>Default Value</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>restart-strategy.failure-rate.max-failures-per-interval</td> 
   <td>失败作业前给定时间间隔内的最大重启次数</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>restart-strategy.failure-rate.failure-rate-interval</td> 
   <td>测量故障率的时间间隔。</td> 
   <td>1 minute</td> 
  </tr> 
  <tr> 
   <td>restart-strategy.failure-rate.delay</td> 
   <td>两次连续重启尝试之间的延迟</td> 
   <td>akka.ask.timeout</td> 
  </tr> 
 </tbody> 
</table>

restart-strategy.failure-rate.max-failures-per-interval: 3
    restart-strategy.failure-rate.failure-rate-interval: 5 min
    restart-strategy.failure-rate.delay: 10 s

故障率重启策略也可以通过程序设置：

val env = ExecutionEnvironment.getExecutionEnvironment()
    env.setRestartStrategy(RestartStrategies.failureRateRestart(
      3, // max failures per unit
      Time.of(5, TimeUnit.MINUTES), //time interval for measuring failure rate
      Time.of(10, TimeUnit.SECONDS) // delay
    ))

### No Restart Strategy ###

作业直接失败，没有尝试重新启动。

restart-strategy: none

无重启策略也可以通过代码设置：

val env = ExecutionEnvironment.getExecutionEnvironment()
    env.setRestartStrategy(RestartStrategies.noRestart())

### Fallback Restart Strategy ###

使用集群定义的重启策略。这对于支持检查点的流程序很有帮助。默认情况下，如果没有定义其他重启策略，则选择固定延迟重启策略。

## 故障恢复策略 ##

Flink 支持不同的故障转移策略，可以通过配置 flink-conf.yaml 中的 *jobmanager.execution.failover-strategy 参数控制。*

<table> 
 <thead> 
  <tr> 
   <th>Failover Strategy</th> 
   <th>Value for jobmanager.execution.failover-strategy</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Restart all</td> 
   <td>full</td> 
  </tr> 
  <tr> 
   <td>Restart pipelined region</td> 
   <td>region</td> 
  </tr> 
 </tbody> 
</table>

### 重新启动所有故障转移策略 ###

此策略重新启动作业中的所有任务以从任务失败中恢复。

### Restart Pipelined Region Failover Strategy ###

这种策略把任务分成互不相连的区域。当检测到任务失败时，此策略计算必须重新启动以从失败中恢复的最小区域集。对于某些作业，与 Restart All Failover 策略相比，这会导致重新启动的任务更少。

区域是一组通过流水线数据交换进行通信的任务。也就是说，批量数据交换表示一个区域的边界：

*  DataStream作业或流表/SQL作业中的所有数据交换都是流水线的。
 *  默认情况下，批处理表/SQL作业中的所有数据交换都是成批处理的。
 *  数据集作业中的数据交换类型由ExecutionMode决定，可以通过ExecutionConfig设置该模式。

重新启动的区域决定如下：

*  包含失败任务的区域将重新启动。
 *  如果结果分区在被重新启动的区域时不可用，产生结果分区的区域也将重新启动。
 *  如果要重新启动某个区域，则该区域的所有消费者区域也将重新启动。这是为了保证数据一致性，因为不确定性处理或分区可能导致不同的分区。

原文地址：[https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/task\_failure\_recovery.html][https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_task_failure_recovery.html]

[https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_task_failure_recovery.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/task_failure_recovery.html