Flink Restart Strategies重啟策略機制深入剖析-Flink牛刀小試
版權宣告:本套技術專欄是作者(秦凱新)平時工作的總結和昇華,通過從真實商業環境抽取案例進行總結和分享,並給出商業應用的調優建議和叢集環境容量規劃等內容,請持續關注本套部落格。版權宣告:禁止轉載,歡迎學習。QQ郵箱地址:[email protected],如有任何問題,可隨時聯絡。
寫在前面的話
為什麼老是寫版權宣告?這裡主要是一些垃圾網站如:碼農網等會爬蟲該網頁內容,卻不標明出處,真是令人厭惡,再次強調,各位讀者朋友勿怪!
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中配置,表示全域性的配置。也可以在應用程式碼中動態指定,會覆蓋全域性配置
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 複製程式碼
-
第二種:應用程式碼設定
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 結語
辛苦成文,各自珍惜,謝謝!
版權宣告:本套技術專欄是作者(秦凱新)平時工作的總結和昇華,通過從真實商業環境抽取案例進行總結和分享,並給出商業應用的調優建議和叢集環境容量規劃等內容,請持續關注本套部落格。版權宣告:禁止轉載,歡迎學習。QQ郵箱地址:[email protected],如有任何問題,可隨時聯絡。
秦凱新 於深圳 20181125