重启策略
Flink支持不同的重启策略,可以控制在发生故障时如何重新启动作业。可以使用默认重新启动策略启动集群,该策略在未定义任何特定于作业的重新启动策略时始终使用。如果使用重新启动策略提交作业,此策略将覆盖群集的默认设置。
概览
默认重启策略是通过Flink的配置文件设置的flink-conf.yaml
。配置参数_restart-strategy_定义采用的策略。如果未启用检查点,则使用“无重启”策略。如果激活了检查点并且尚未配置重启策略,则固定延迟策略将用于 Integer.MAX_VALUE
重启尝试。请参阅以下可用的重新启动策略列表,以了解支持的值。
每个重启策略都有自己的一组参数来控制其行为。这些值也在配置文件中设置。每个重新启动策略的描述包含有关相应配置值的更多信息。
重启策略 | 重启策略的Value |
---|---|
固定延迟 | 固定延迟 |
失败率 | 失败率 |
不重启 | no / not |
除了定义默认重启策略之外,还可以为每个Flink作业定义特定的重启策略。通过调用setRestartStrategy
on上的方法以编程方式设置此重新启动策略ExecutionEnvironment
。请注意,这也适用于StreamExecutionEnvironment
。
以下示例显示了我们如何为我们的工作设置固定延迟重启策略。如果发生故障,系统会尝试重新启动作业3次,并在连续重启尝试之间等待10秒。
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // number of restart attempts
Time.of(10, TimeUnit.SECONDS) // delay
));
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // number of restart attempts
Time.of(10, TimeUnit.SECONDS) // delay ))
重启策略
以下部分介绍重新启动策略特定的配置选项。
固定延迟重启策略
固定延迟重启策略尝试给定次数重新启动作业。如果超过最大尝试次数,则作业最终会失败。在两次连续重启尝试之间,重启策略等待一段固定的时间。
通过在中设置以下配置参数,此策略默认启用flink-conf.yaml
。
restart-strategy: fixed-delay
配置参数 | 描述 | 默认值 |
---|---|---|
restart-strategy.fixed-delay.attempts |
Flink在作业声明失败之前重试执行的次数。 | 1,或者Integer.MAX_VALUE 如果通过检查点激活 |
restart-strategy.fixed-delay.delay |
延迟重试意味着在执行失败后,重新执行不会立即开始,而是仅在一定延迟之后。当程序与外部系统交互时,延迟重试可能会有所帮助,例如,在尝试重新执行之前,连接或挂起的事务应该达到超时。 | akka.ask.timeout ,如果通过检查点激活,则为10秒 |
例如:
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s
固定延迟重启策略也可以通过编程方式设置:
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // number of restart attempts
Time.of(10, TimeUnit.SECONDS) // delay
));
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // number of restart attempts
Time.of(10, TimeUnit.SECONDS) // delay ))
故障率重启策略
故障率重启策略在故障后重新启动作业,但是当failure rate
超过(每个时间间隔的故障)时,作业最终会失败。在两次连续重启尝试之间,重启策略等待一段固定的时间。
通过在中设置以下配置参数,此策略默认启用flink-conf.yaml
。
restart-strategy: failure-rate
配置参数 | 描述 | 默认值 |
---|---|---|
<it>重新启动-strategy.failure-rate.max-故障每间隔</it> | 失败作业之前给定时间间隔内的最大重启次数 | 1 |
<it>重启strategy.failure-rate.failure速率间隔</it> | 测量故障率的时间间隔。 | 1分钟 |
<it>重启strategy.failure-rate.delay</it> | 两次连续重启尝试之间的延迟 | <it>akka.ask.timeout</it> |
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
故障率重启策略也可以通过编程方式设置:
ExecutionEnvironment env = ExecutionEnvironment.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
));
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 ))
不重启策略
作业直接失败,不尝试重启。
restart-strategy: none
也可以通过编程方式设置no restart策略:
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.noRestart())
回退重启策略
使用群集定义的重新启动策略。这对于启用检查点的流式传输程序很有帮助。默认情况下,如果没有定义其他重启策略,则选择固定延迟重启策略。