[case49]聊聊flink的checkpoint配置

序

本文主要研究下flink的checkpoint配置

實例

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// start a checkpoint every 1000 ms
env.enableCheckpointing(1000);

// advanced options:

// set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig().setCheckpointTimeout(60000);

// make sure 500 ms of progress happen between checkpoints
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// enable externalized checkpoints which are retained after job cancellation
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// This determines if a task will be failed if an error occurs in the execution of the task’s checkpoint procedure.
env.getCheckpointConfig().setFailOnCheckpointingErrors(true);

• 使用StreamExecutionEnvironment.enableCheckpointing方法來設置開啓checkpoint；具體能夠使用enableCheckpointing(long interval)，或者enableCheckpointing(long interval, CheckpointingMode mode)；interval用於指定checkpoint的觸發間隔(單位milliseconds)，而CheckpointingMode默認是CheckpointingMode.EXACTLY_ONCE，也能夠指定爲CheckpointingMode.AT_LEAST_ONCE
• 也能夠經過StreamExecutionEnvironment.getCheckpointConfig().setCheckpointingMode來設置CheckpointingMode，通常對於超低延遲的應用(大概幾毫秒)能夠使用CheckpointingMode.AT_LEAST_ONCE，其餘大部分應用使用CheckpointingMode.EXACTLY_ONCE就能夠
• checkpointTimeout用於指定checkpoint執行的超時時間(單位milliseconds)，超時沒完成就會被abort掉
• minPauseBetweenCheckpoints用於指定checkpoint coordinator上一個checkpoint完成以後最小等多久能夠出發另外一個checkpoint，當指定這個參數時，maxConcurrentCheckpoints的值爲1
• maxConcurrentCheckpoints用於指定運行中的checkpoint最多能夠有多少個，用於包裝topology不會花太多的時間在checkpoints上面；若是有設置了minPauseBetweenCheckpoints，則maxConcurrentCheckpoints這個參數就不起做用了(大於1的值不起做用)
• enableExternalizedCheckpoints用於開啓checkpoints的外部持久化，可是在job失敗的時候不會自動清理，須要本身手工清理state；ExternalizedCheckpointCleanup用於指定當job canceled的時候externalized checkpoint該如何清理，DELETE_ON_CANCELLATION的話，在job canceled的時候會自動刪除externalized state，可是若是是FAILED的狀態則會保留；RETAIN_ON_CANCELLATION則在job canceled的時候會保留externalized checkpoint state
• failOnCheckpointingErrors用於指定在checkpoint發生異常的時候，是否應該fail該task，默認爲true，若是設置爲false，則task會拒絕checkpoint而後繼續運行

flink-conf.yaml相關配置

#==============================================================================
# Fault tolerance and checkpointing
#==============================================================================

# The backend that will be used to store operator state checkpoints if
# checkpointing is enabled.
#
# Supported backends are 'jobmanager', 'filesystem', 'rocksdb', or the
# <class-name-of-factory>.
#
# state.backend: filesystem

# Directory for checkpoints filesystem, when using any of the default bundled
# state backends.
#
# state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints

# Default target directory for savepoints, optional.
#
# state.savepoints.dir: hdfs://namenode-host:port/flink-checkpoints

# Flag to enable/disable incremental checkpoints for backends that
# support incremental checkpoints (like the RocksDB state backend). 
#
# state.backend.incremental: false

• state.backend用於指定checkpoint state存儲的backend，默認爲none
• state.backend.async用於指定backend是否使用異步snapshot(默認爲true)，有些不支持async或者只支持async的state backend可能會忽略這個參數
• state.backend.fs.memory-threshold，默認爲1024，用於指定存儲於files的state大小閾值，若是小於該值則會存儲在root checkpoint metadata file
• state.backend.incremental，默認爲false，用於指定是否採用增量checkpoint，有些不支持增量checkpoint的backend會忽略該配置
• state.backend.local-recovery，默認爲false
• state.checkpoints.dir，默認爲none，用於指定checkpoint的data files和meta data存儲的目錄，該目錄必須對全部參與的TaskManagers及JobManagers可見
• state.checkpoints.num-retained，默認爲1，用於指定保留的已完成的checkpoints個數
• state.savepoints.dir，默認爲none，用於指定savepoints的默認目錄
• taskmanager.state.local.root-dirs，默認爲none

小結

• 能夠經過使用StreamExecutionEnvironment.enableCheckpointing方法來設置開啓checkpoint；具體能夠使用enableCheckpointing(long interval)，或者enableCheckpointing(long interval, CheckpointingMode mode)
• checkpoint的高級配置能夠配置checkpointTimeout(用於指定checkpoint執行的超時時間，單位milliseconds)，minPauseBetweenCheckpoints(用於指定checkpoint coordinator上一個checkpoint完成以後最小等多久能夠出發另外一個checkpoint)，maxConcurrentCheckpoints(用於指定運行中的checkpoint最多能夠有多少個，若是有設置了minPauseBetweenCheckpoints，則maxConcurrentCheckpoints這個參數大於1的值不起做用)，enableExternalizedCheckpoints(用於開啓checkpoints的外部持久化，在job failed的時候externalized checkpoint state沒法自動清理，可是在job canceled的時候能夠配置是刪除仍是保留state)
• 在flink-conf.yaml裏頭也有checkpoint的相關配置，主要是state backend的配置，好比state.backend.async、state.backend.incremental、state.checkpoints.dir、state.savepoints.dir等

doc

• Checkpointing