当前位置：网站首页>Flink restart policy

Flink restart policy

2022-07-24 06:16:00 【sf_ www】

When the task fails ,Flink You need to restart the failed tasks and other affected tasks , To restore the task to normal .
Restart strategy (Restart strategies) And failover strategies (failover strategies) Used to control task restart . Restart strategy determines whether and when restart fails / Affected tasks . The failover strategy determines which tasks should be restarted to resume the job .

The cluster can be started using the default restart policy , When there is no job specific restart policy defined , Always use the default restart policy . If you use the restart policy to submit jobs , Then this policy will override the default settings of the cluster .
The default restart policy is through Flink Configuration file for link-conf.yaml Set up , Configuration parameters restart-strategy Defines which strategy to use . If checkpoints are not enabled , Then use " No restart " Strategy . If the checkpoint is activated , And the restart policy is not configured , The fixed delay strategy is used （ Restart attempts use Integer.MAX_VALUE）.

Except in the flink-conf.yaml Set the default restart policy in , We can also do it for each Flink Job definition specific restart strategy . This restart strategy is achieved by calling StreamExecutionEnvironment Upper setRestartStrategy Method is programmed .

The current restart strategy is as follows 5 Kind of ：

（ You can see org.apache.flink.api.common.restartstrategy.RestartStrategies class ）

1） Fixed delay strategy （Fixed Delay Restart Strategy）
2） Exponential delay strategy （Exponential Delay Restart Strategy）
3） Failure rate restart strategy （Failure Rate Restart Strategy）
4） No restart policy （No Restart Strategy）
5） Backup strategy （Fallback Restart Strategy）

1. Fixed delay strategy （Fixed Delay Restart Strategy）

The fixed delay restart strategy is to try to restart the job for a given number of times . If the maximum number of attempts is exceeded , The mission will eventually fail . Between two consecutive restart attempts , The restart strategy needs to wait for a fixed time .

stay flink-conf.yaml Set in ：

#  Restart strategy 
restart-strategy: fixed-delay
#  Number of attempts 
restart-strategy.fixed-delay.attempts: 3
#  Fixed delay time 
restart-strategy.fixed-delay.delay: 10 s

Set... In the code ：

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

2. Exponential delay strategy （Exponential Delay Restart Strategy）

The exponential delay restart strategy attempts to restart the job indefinitely , As the delay increases , Until the maximum delay . Tasks never fail . Between two consecutive restart attempts , The waiting time of restart strategy will increase exponentially , Until the maximum . then , It keeps the waiting delay time at the maximum . If the job is performed correctly , Exponential delay value after a period of time （restart-strategy.exponential-delay.reset-backoff-threshold） Reset to the original value （restart-strategy.exponential-delay.initial-backoff）.

stay flink-conf.yaml Set in ：

#  Restart strategy 
restart-strategy: exponential-delay
#  Restart interval after initial failure （ Initial value ）
restart-strategy.exponential-delay.initial-backoff: 10 s
#  Maximum restart interval , After exceeding this maximum , The restart interval is no longer increased 
restart-strategy.exponential-delay.max-backoff: 2 min
#  After each failure , The restart interval is the last restart interval multiplied by this value 
restart-strategy.exponential-delay.backoff-multiplier: 2.0
#  How long does the job run without failure , The restart interval will be reset to the initial value （ The value of the first configuration item ）
restart-strategy.exponential-delay.reset-backoff-threshold: 10 min
#  The maximum jitter value of each restart interval （ Add or subtract a random number within the scope of the configuration item ）, Prevent a large number of jobs from restarting at the same time 
restart-strategy.exponential-delay.jitter-factor: 0.1

Set... In the code ：

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 （Failure Rate Restart Strategy）

The failure rate restart strategy is to restart the task after the task fails , But when the failure rate ( Failure rate of each time interval ) Over time , The mission will eventually fail . Between two consecutive restart attempts , The restart strategy needs to wait for a certain time . That is to say restart-strategy.failure-rate.failure-rate-interval Failed for more than restart-strategy.failure-rate.max-failures-per-interval This value fails .

stay flink-conf.yaml Set in ：

#  Restart strategy 
restart-strategy: failure-rate
#  Maximum number of restarts in a given time interval before a failed job 
restart-strategy.failure-rate.max-failures-per-interval: 3
#  Time interval for measuring failure rate 
restart-strategy.failure-rate.failure-rate-interval: 5 min
#  Delay between two consecutive restart attempts 
restart-strategy.failure-rate.delay: 10 s

Set... In the code ：

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 policy （No Restart Strategy）

The job failed directly , Don't try to restart .

stay flink-conf.yaml Set in ：

restart-strategy: none

Set... In the code ：

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

5. Backup strategy （Fallback Restart Strategy）

Use the restart strategy defined by the cluster . This is very helpful for streaming programs that support checkpointing . By default , If no other restart policy is defined , Select the fixed delay restart policy .

Applicable scenarios, especially when you have a custom restart strategy implementation , stay flink-conf.yaml Especially useful when configuring .