版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。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