聊聊flink的checkpoint配置
阿新 • • 發佈:2018-12-09
序
本文主要研究下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等