最具影响力的数字化技术在线社区

168大数据

 找回密码
 立即注册

QQ登录

只需一步,快速开始

1 2 3 4 5
开启左侧

[Flink] Flink重启策略机制

[复制链接]
发表于 2021-12-7 19:43:55 | 显示全部楼层 |阅读模式

马上注册,结交更多数据大咖,获取更多知识干货,轻松玩转大数据

您需要 登录 才可以下载或查看,没有帐号?立即注册

x
1、Failure Rate Restart Strategy 说明
故障率重启策略,flink提供的一种比较"智能"的重启策略;即当任务的失败率上升到一定的程度时,flink认为本次任务最终是失败的;
也可以理解为,在该策略中,flink关注的点是任务的失败率,失败率计算公式如下:
失败率 = 失败次数/时间区间
“失败次数”对restart-strategy.failure-rate.max-failures-per-interval,
“时间区间”对应的配置项为restart-strategy.failure-rate.failure-rate-interval,也就是说在该时间范围内,
运行时失败次数超过配置的失败次数则任务最终失败;当然,类似于Fixed Delay Restart Strategy,在每两次连续的重试之间也会有一个固定时间的delay。

2、文件配置说明
①失败率配置
[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  #间隔


3、代码配置
①失败率配置
[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) // 间隔
));


4、无重启配置
在特殊情况下,希望处理出错就不重复执行,直接报出错,配置如下:
配置文件:
restart-strategy: none
代码配置:
[AppleScript] 纯文本查看 复制代码
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());


5、代码示例
[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);




楼主热帖
168大数据(www.bi168.cn)是国内首家系统性关注大数据科学与人工智能的社区媒体!
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

站长推荐上一条 /1 下一条

关于我们|小黑屋|Archiver|168大数据 ( 京ICP备14035423号|申请友情链接

GMT+8, 2022-1-21 17:07 , Processed in 0.070350 second(s), 17 queries , Xcache On.

Powered by BI168大数据社区

© 2012-2014 168大数据

快速回复 返回顶部 返回列表