转载自:https://blog.csdn.net/shenshouniu/article/details/84558874
欢迎加入大数据学习群:
**Flink学习视频:**http://edu.51cto.com/sd/88e07
1 重启策略总览
-
Flink支持不同的重启策略,以在故障发生时控制作业如何重启
集群在启动时会伴随一个默认的重启策略,在没有定义具体重启策略时会使用该默认策略。 如果在工作提交时指定了一个重启策略,该策略会覆盖集群的默认策略 -
默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。
-
常用的重启策略
固定间隔 (Fixed delay)
失败率 (Failure rate)
无重启 (No restart)
-
如果没有启用 checkpointing,则使用无重启 (no restart) 策略。
-
如果启用了 checkpointing,但没有配置重启策略,则使用固定间隔 (fixed-delay) 策略,其中 Integer.MAX_VALUE 参数是尝试重启次数
重启策略可以在flink-conf.yaml中配置,表示全局的配置。也可以在应用代码中动态指定,会覆盖全局配置扫描二维码关注公众号,回复: 5016181 查看本文章
2 重启策略之固定间隔
-
第一种:全局配置 flink-conf.yaml
restart-strategy: fixed-delay restart-strategy.fixed-delay.attempts: 3 restart-strategy.fixed-delay.delay: 10 s
第二种:应用代码设置
env.setRestartStrategy(RestartStrategies.fixedDelayRestart( // 尝试重启的次数 Time.of(10, TimeUnit.SECONDS) // 间隔 ));
3 重启策略之失败率
-
失败率重启策略在Job失败后会重启,但是超过失败率后,Job会最终被认定失败。在两个连续的重启尝试之间,重启策略会等待一个固定的时间.
-
如下配置是5分钟内若失败了3次则认为该job失败,重试间隔为10s。
-
第一种:全局配置 flink-conf.yaml
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
-
第二种:应用代码设置
env.setRestartStrategy(RestartStrategies.failureRateRestart(
//一个时间段内的最大失败次数
Time.of(5, TimeUnit.MINUTES), // 衡量失败次数的是时间段
Time.of(10, TimeUnit.SECONDS) // 间隔
));
4 无重启策略
无重启策略,Job直接失败,不会尝试进行重启
-
第一种:全局配置 flink-conf.yaml
restart-strategy: none
- 1
第二种:应用代码设置
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());
5 案例代码实战
-
如果启用了 checkpointing,但没有配置重启策略,则使用固定间隔 (fixed-delay) 策略
//获取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);
//设置statebackend
//env.setStateBackend(new MemoryStateBackend());
//env.setStateBackend(new FsStateBackend(“hdfs://hadoop100:9000/flink/checkpoints”));
//env.setStateBackend(new RocksDBStateBackend(“hdfs://hadoop100:9000/flink/checkpoints”,true));
6 结语
</div>