flink任务失败restart策略和failover策略

flink任务失败restart策略和failover策略 一、restart策略

Flink在任务失败恢复重启时支持多种重启策略，同时配置支持两个级别，可以通过设置flink全局配置文件flink-conf.yaml来指定全局重启策略，也可以通过编程的方式指定单个job的重启策略，极端情况下可以为每一个任务都指定一个重启策略；一般情况下，编程方式指定的单个job重启策略会将全局策略覆盖。

在检查点机制未开启的情况下，flink默认会采取No Restart Strategy策略，即失败不重启；在检查点机制开启状态下，默认会采取Fixed Delay Restart Strategy（固定间隔延迟重启）策略，且重试次数为Integer.MAX_VALUE。

1、Fixed Delay Restart Strategy

固定间隔延迟重启策略，表现形式为：给定延迟间隔，给定重试次数，且两次连续重试尝试之间必须保证等待配置的固定时间，在重试了给定次数之后依然失败，那么将最终失败。

1）配置方式设置： 

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

2）编程方式设置：

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

2、Exponential Delay Restart Strategy

指数延迟重启策略尝试无限重启作业，将延迟增加到最大延迟。 这项工作永远不会失败。 在两次连续的重启尝试之间，重启策略保持指数增长，直到达到最大次数。 然后，它将延迟保持在最大数量。

当作业正确执行时，指数延迟值会在一段时间后重置； 这个阈值是可配置的。

1）配置方式设置：

restart-strategy: exponential-delay
restart-strategy.exponential-delay.initial-backoff: 10 s
restart-strategy.exponential-delay.max-backoff: 2 min
restart-strategy.exponential-delay.backoff-multiplier: 2.0
restart-strategy.exponential-delay.reset-backoff-threshold: 10 min
restart-strategy.exponential-delay.jitter-factor: 0.1

2）代码方式设置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.exponentialDelayRestart(
  Time.milliseconds(1),
  Time.milliseconds(1000),
  1.1, // exponential multiplier
  Time.milliseconds(2000), // threshold duration to reset delay to its initial value
  0.1 // jitter
));

3、Failure Rate Restart Strategy

故障率重启策略，flink提供的一种比较"智能"的重启策略；即当任务的失败率上升到一定的程度时，flink认为本次任务最终是失败的；也可以理解为，在该策略中，flink关注的点是任务的失败率，那么具体生产中如何配置该策略的失败率就是我们需要根据实际的情况去权衡选择了。

失败率计算公式：失败率 = 失败次数/时间区间

1）配置方式设置：

restart-strategy: failure-rate
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

2）代码方式设置：

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

4、No Restart Strategy 

任务执行失败后，不会尝试重启，直接失败；如果Flink的检查点机制没开启的话，全局默认重启策略就是No Restart Strategy；很好理解，因为没了恢复依赖的分布式快照，无法对任务进行恢复。

1）配置方式设置：

restart-strategy: none

2）代码方式设置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());

5、Fallback Restart Strategy
后备重启策略，保底策略，即所谓的默认重启策略；集群中如果没有在配置文件（flink-conf.yaml）中显示的配置重启策略，也没有在编程中织入，在检查点机制开启的情况下，任务失败，flink会默认的选用Fixed Delay Restart Strategy重启，且会无限尝试重连（Integer.MAX_VALUE次）。

二、failover策略

可以在flink-config.yml中设置jobmanager.execution.failover-strategy 来确定failover策略：

• full：restart all，This strategy restarts all tasks in the job to recover from a task failure.
• region：Restart pipelined region        

region failover策略：

该策略将任务分组到不相交的区域中。当检测到任务失败时，此策略计算必须重新启动才能从失败中恢复的最小区域集。对于某些作业，与重新启动所有故障转移策略相比，这可能会导致重新启动的任务更少。

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

• All data exchanges in a DataStream job or Streaming Table/SQL job are pipelined.
• All data exchanges in a Batch Table/SQL job are batched by default.
• The data exchange types in a DataSet job are determined by the ExecutionMode which can be set through ExecutionConfig.

重启区域决定如下：

• 包含失败任务的区域将重新启动。
• 如果结果分区不可用而将重新启动的区域需要它，则生成结果分区的区域也将重新启动。
• 如果一个区域要重启，它的所有消费者区域也将被重启。这是为了保证数据一致性，因为非确定性处理或分区会导致不同的分区。

官方文档：https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/task_failure_recovery/