Flink的重启策略
Flink的重啟策略
Flink支持不同的重啟策略,這些重啟策略控制著job失敗后如何重啟。集群可以通過默認(rèn)的重啟策略來重啟,這個(gè)默認(rèn)的重啟策略通常在未指定重啟策略的情況下使用,而如果Job提交的時(shí)候指定了重啟策略,這個(gè)重啟策略就會(huì)覆蓋掉集群的默認(rèn)重啟策略。
概覽
集群在啟動(dòng)時(shí)會(huì)伴隨一個(gè)默認(rèn)的重啟策略,在沒有定義具體重啟策略時(shí),會(huì)使用該默認(rèn)重啟策略,如果在工作提交時(shí)指定了一個(gè)重啟策略,那么該策略會(huì)覆蓋集群的默認(rèn)策略。
默認(rèn)的重啟策略可以通過Flink的配置文件flink-conf.yaml指定,配置參數(shù)restart-strategy定義了哪個(gè)策略被使用。
常用的重啟策略:
- 固定間隔(Fixed delay)
- 失敗率(Failure rate)
- 無重啟(No restart)
請(qǐng)參考下面的可用重啟策略來了解哪些值是支持的。
每個(gè)重啟策略都有自己的參數(shù)來控制它的行為,這些值也可以在配置文件中設(shè)置,每個(gè)重啟策略的描述都包含著各自的配置值信息。
| Fixed delay | fixed-delay |
| Failure rate | failure-rate |
| No restart | None |
除了定義一個(gè)默認(rèn)的重啟策略之外,你還可以為每一個(gè)Job指定它自己的重啟策略,這個(gè)重啟策略可以在ExecutionEnvironment中調(diào)用setRestartStrategy()方法來程序化地調(diào)用,主意這種方式同樣適用于StreamExecutionEnvironment。
實(shí)例
下面的例子展示了我們?nèi)绾螢槲覀兊腏ob設(shè)置一個(gè)固定延遲重啟策略,一旦有失敗,系統(tǒng)就會(huì)嘗試每10秒重啟一次,重啟3次。
java方式
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, // 嘗試重啟次數(shù)Time.of(10, TimeUnit.SECONDS) // 延遲時(shí)間間隔 ));scala方式
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, // 重啟次數(shù)Time.of(10, TimeUnit.SECONDS) // 延遲時(shí)間間隔 ))重啟策略
下面部分描述了重啟策略特定的配置項(xiàng)
固定延遲重啟策略(Fixed Delay Restart Strategy)
固定延遲重啟策略會(huì)嘗試一個(gè)給定的次數(shù)來重啟Job,如果超過了最大的重啟次數(shù),Job最終將失敗。在連續(xù)的兩次重啟嘗試之間,重啟策略會(huì)等待一個(gè)固定的時(shí)間。
重啟策略可以配置flink-conf.yaml的下面配置參數(shù)來啟用,作為默認(rèn)的重啟策略:
| restart-strategy.fixed-delay.attempts | 在Job最終宣告失敗之前,Flink嘗試執(zhí)行的次數(shù) | 1,如果啟用checkpoint的話是Integer.MAX_VALUE |
| restart-strategy.fixed-delay.delay | 延遲重啟意味著一個(gè)執(zhí)行失敗之后,并不會(huì)立即重啟,而是要等待一段時(shí)間。 | akka.ask.timeout,如果啟用checkpoint的話是1s |
例子:
restart-strategy.fixed-delay.attempts: 3 restart-strategy.fixed-delay.delay: 10 s固定延遲重啟也可以在程序中設(shè)置:
java:
scala:
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, // 重啟次數(shù)Time.of(10, TimeUnit.SECONDS) // 重啟時(shí)間間隔 ))失敗率重啟策略(Failure rate)
失敗率重啟策略在Job失敗后會(huì)重啟,但是超過失敗率后,Job會(huì)最終被認(rèn)定失敗。在兩個(gè)連續(xù)的重啟嘗試之間,重啟策略會(huì)等待一個(gè)固定的時(shí)間。
失敗率重啟策略可以在flink-conf.yaml中設(shè)置下面的配置參數(shù)來啟用:
| restart-strategy.failure-rate.max-failures-per-interval | 在一個(gè)Job認(rèn)定為失敗之前,最大的重啟次數(shù) | 1 |
| restart-strategy.failure-rate.failure-rate-interval | 計(jì)算失敗率的時(shí)間間隔 | 1分鐘 |
| restart-strategy.failure-rate.delay | 兩次連續(xù)重啟嘗試之間的時(shí)間間隔 | akka.ask.timeout |
例子:
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失敗率重啟策略也可以在程序中設(shè)置:
Java代碼:
Scala代碼::
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.failureRateRestart(3, // 每個(gè)測(cè)量時(shí)間間隔最大失敗次數(shù)Time.of(5, TimeUnit.MINUTES), //失敗率測(cè)量的時(shí)間間隔Time.of(10, TimeUnit.SECONDS) // 兩次連續(xù)重啟嘗試的時(shí)間間隔 ))無重啟策略
Job直接失敗,不會(huì)嘗試進(jìn)行重啟。如果沒有啟動(dòng)checkpoint,則默認(rèn)情況下就是無重啟
restart-strategy: none無重啟策略也可以在程序中設(shè)置
Java代碼:
Scala代碼:
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.noRestart())總結(jié)
以上是生活随笔為你收集整理的Flink的重启策略的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Flink 广播变量
- 下一篇: Flink countWindow窗口