Flink状态与容错

一致性检查点

1.什么是一致性检查点

Flink故障恢复机制的核心，就是应用的一致性检查点。有状态应用的一致性检查点，就是所有任务的状态，在某个时间点的一份快照。这个时间点，是所有任务都恰好处理完同一个相同输入数据的时候。

在流式应用处理过程中，Flink会周期性的为应用状态生成检查点。一旦发生故障，Flink会利用最新的检查点来恢复任务状态，并重新启动处理程序。如果所有算子的状态都可以写入检查点并从其中恢复，并且所有输入流的消费位置都能重置到检查点的那一刻，那么检查点和故障恢复机制就能为整个应用状态提供精确一次的一致性保障。 值得一提的是，Flink的检查点机制，只能保证应用内部状态的精确一致性，至于端对端的精确一致性保证，还需结合外部读写系统的支持。

val env = StreamExecutionEnvironment.getExecutionEnvironment()

// 启用checkpoint间隔 1000ms
env.enableCheckpointing(1000)

// 高级配置:

// 设置 exactly-once (this is the default)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

// 设置checkpoint最小间隔 500ms
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

// 设置checkpoint在1分钟内完成 否则将被丢弃
env.getCheckpointConfig.setCheckpointTimeout(60000)

// 设置 checkpoint 失败时，任务不会 fail，该 checkpoint 会被丢弃
env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)

// 设置checkpoint并发度为1
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

// 任务canel后，检查点的状态
CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);