Flink重启策略机制 - 168大数据

故障率重启策略，flink提供的一种比较"智能"的重启策略；即当任务的失败率上升到一定的程度时，flink认为本次任务最终是失败的；

也可以理解为，在该策略中，flink关注的点是任务的失败率，失败率计算公式如下：

失败率 = 失败次数/时间区间
“失败次数”对restart-strategy.failure-rate.max-failures-per-interval，

“时间区间”对应的配置项为restart-strategy.failure-rate.failure-rate-interval，也就是说在该时间范围内，

运行时失败次数超过配置的失败次数则任务最终失败；当然，类似于Fixed Delay Restart Strategy，在每两次连续的重试之间也会有一个固定时间的delay。

[AppleScript] 纯文本查看 复制代码

restart-strategy: failure-rate

restart-strategy.failure-rate.max-failures-per-interval: 3  #失败的次数

restart-strategy.failure-rate.failure-rate-interval: 2 min  #时间段内

restart-strategy.failure-rate.delay: 5s  #重试的间隔时间



间段内任务失败后，重启时，每两次连续尝试之间间隔5秒，在2分钟内任务重启失败3次，则任务最终失败，配置默认全局生效。②间隔时间配置



restart-strategy: fixed-delay

restart-strategy.fixed-delay.attempts: 3  #次数

restart-strategy.fixed-delay.delay: 10 s  #间隔

[AppleScript] 纯文本查看 复制代码

env.setRestartStrategy(RestartStrategies.failureRateRestart(

3,//最大失败次数

Time.of(2, TimeUnit.MINUTES), // 衡量失败次数的是时间段

Time.of(5, TimeUnit.SECONDS) // 间隔

));

②时间间隔配置

[AppleScript] 纯文本查看 复制代码

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(

  3,,// 尝试重启的次数

  Time.of(5, TimeUnit.SECONDS) // 间隔

));

在特殊情况下，希望处理出错就不重复执行，直接报出错，配置如下：

restart-strategy: none

[AppleScript] 纯文本查看 复制代码

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

env.setRestartStrategy(RestartStrategies.noRestart());

[AppleScript] 纯文本查看 复制代码

//获取flink的运行环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

 

// 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】

env.enableCheckpointing(1000);

// 高级选项：

// 设置模式为exactly-once （这是默认值）

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】

env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】

env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同一时间只允许进行一个检查点

env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】

//ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint

//ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint

env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);