1. 程式人生 > >Flink Restart Strategies重啟策略機制深入剖析-Flink牛刀小試

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