• Flink 支持不同的重启策略,可以在我们的Flink的Job发生故障的时候控制作业的重启机制。
  • 集群在启动时会伴随一个默认的重启策略,在没有定义的具体的重启策略时会使用该默认策略。
  • 重启策略分为全局配置和当前应用程序配置

重启策略的配置

全局配置

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

配置在5分钟内若失败3次则认定作业失败,重试间隔为10s

1
2
3
4
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

当前应用程序配置

1
2
3
4
env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 
3,// 尝试重启的次数
Time.of(10, TimeUnit.SECONDS) // 间隔
));

配置在5分钟内若失败3次则认定作业失败,重试间隔为10s

1
2
3
4
5
 env.setRestartStrategy(RestartStrategies.failureRateRestart(  
3,//一个时间段内的最大失败次数
Time.of(5, TimeUnit.MINUTES), // 衡量失败次数的是时间段
Time.of(10, TimeUnit.SECONDS) // 间隔
));

实际代码的演示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
      //获取flink的运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每隔 1000 ms进行启动一个检查点【设置checkpoint的周期】
// 这个只对流有效
env.enableCheckpointing(1000);
// 间隔10秒 重启3次
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3,
Time.seconds(10)
));
//5分钟内若失败了3次则认为该job失败,重试间隔为10s
env.setRestartStrategy(RestartStrategies.failureRateRestart(
3,
Time.of(5,TimeUnit.MINUTES),
Time.of(10,TimeUnit.SECONDS)
));
//不重试
env.setRestartStrategy(RestartStrategies.noRestart());