Flink的重啟策略
Flink支持不同的重啟策略，這些重啟策略控制著job失敗后如何重啟。集群可以通過默認的重啟策略來重啟，這個默認的重啟策略通常在未指定重啟策略的情況下使用，而如果Job提交的時候指定了重啟策略，這個重啟策略就會覆蓋掉集群的默認重啟策略。
概覽
默認的重啟策略是通過Flink的flink-conf.yaml來指定的，這個配置參數(shù)restart-strategy定義了哪種策略會被采用。如果checkpoint未啟動，就會采用no restart策略，如果啟動了checkpoint機制，但是未指定重啟策略的話，就會采用fixed-delay策略，重試Integer.MAX_VALUE次。請參考下面的可用重啟策略來了解哪些值是支持的。
每個重啟策略都有自己的參數(shù)來控制它的行為，這些值也可以在配置文件中設置，每個重啟策略的描述都包含著各自的配置值信息。
| 重啟策略 | 重啟策略值 |
|---|---|
| Fixed delay | fixed-delay |
| Failure rate | failure-rate |
| No restart | None |
除了定義一個默認的重啟策略之外，你還可以為每一個Job指定它自己的重啟策略，這個重啟策略可以在ExecutionEnvironment中調用setRestartStrategy()方法來程序化地調用，主意這種方式同樣適用于StreamExecutionEnvironment。
下面的例子展示了我們?nèi)绾螢槲覀兊腏ob設置一個固定延遲重啟策略，一旦有失敗，系統(tǒng)就會嘗試每10秒重啟一次，重啟3次。
Java代碼
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 嘗試重啟次數(shù)
Time.of(10, TimeUnit.SECONDS) // 延遲時間間隔
));
Scala代碼:
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 重啟次數(shù)
Time.of(10, TimeUnit.SECONDS) // 延遲時間間隔
))
重啟策略
下面部分描述了重啟策略特定的配置項
固定延遲重啟策略(Fixed Delay Restart Strategy)
固定延遲重啟策略會嘗試一個給定的次數(shù)來重啟Job，如果超過了最大的重啟次數(shù)，Job最終將失敗。在連續(xù)的兩次重啟嘗試之間，重啟策略會等待一個固定的時間。
重啟策略可以配置flink-conf.yaml的下面配置參數(shù)來啟用，作為默認的重啟策略:
restart-strategy: fixed-delay
| 配置參數(shù) | 描述 | 默認值 |
|---|---|---|
| restart-strategy.fixed-delay.attempts | 在Job最終宣告失敗之前，F(xiàn)link嘗試執(zhí)行的次數(shù) | 1，如果啟用checkpoint的話是Integer.MAX_VALUE |
| restart-strategy.fixed-delay.delay | 延遲重啟意味著一個執(zhí)行失敗之后，并不會立即重啟，而是要等待一段時間。 | akka.ask.timeout,如果啟用checkpoint的話是1s |
例子:
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s
固定延遲重啟也可以在程序中設置:
Java代碼:
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 重啟次數(shù)
Time.of(10, TimeUnit.SECONDS) // 重啟時間間隔
));
Scala代碼:
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 重啟次數(shù)
Time.of(10, TimeUnit.SECONDS) // 重啟時間間隔
))
失敗率重啟策略
失敗率重啟策略在Job失敗后會重啟，但是超過失敗率后，Job會最終被認定失敗。在兩個連續(xù)的重啟嘗試之間，重啟策略會等待一個固定的時間。
失敗率重啟策略可以在flink-conf.yaml中設置下面的配置參數(shù)來啟用:
restart-strategy:failure-rate
| 配置參數(shù) | 描述 | 默認值 |
|---|---|---|
| restart-strategy.failure-rate.max-failures-per-interval | 在一個Job認定為失敗之前，最大的重啟次數(shù) | 1 |
| restart-strategy.failure-rate.failure-rate-interval | 計算失敗率的時間間隔 | 1分鐘 |
| restart-strategy.failure-rate.delay | 兩次連續(xù)重啟嘗試之間的時間間隔 | 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
失敗率重啟策略也可以在程序中設置:
Java代碼:
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.failureRateRestart(
3, // 每個測量時間間隔最大失敗次數(shù)
Time.of(5, TimeUnit.MINUTES), //失敗率測量的時間間隔
Time.of(10, TimeUnit.SECONDS) // 兩次連續(xù)重啟嘗試的時間間隔
));
Scala代碼:
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.failureRateRestart(
3, // 每個測量時間間隔最大失敗次數(shù)
Time.of(5, TimeUnit.MINUTES), //失敗率測量的時間間隔
Time.of(10, TimeUnit.SECONDS) // 兩次連續(xù)重啟嘗試的時間間隔
))
無重啟策略
Job直接失敗，不會嘗試進行重啟
restart-strategy: none
無重啟策略也可以在程序中設置
Java代碼:
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());
Scala代碼:
val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.noRestart())
