flinkFlink专题

Flink Streaming检查点(Checkpointing

2019-02-11  本文已影响213人  尼小摩

Flink中的每个函数和操作符都可以是有状态的(有关详细信息,请参见使用状态)。有状态函数在单个元素/事件的处理过程中存储数据,使状态成为任何类型的更精细操作的关键构建块。

为了使状态容错,Flink需要对状态进行checkpoint(检查点)。检查点允许Flink恢复流中的状态和位置,从而为应用程序提供与无故障执行相同的语义。
documentation on streaming fault tolerance详细描述了Flink流容错机制背后的技术。

前提

Flink的检查点机制与流和状态的持久存储交互。一般而言,它要求:

启用和配置检查点(Enabling and Configuring Checkpointing)

默认情况下,禁用检查点。要启用检查点请在StreamExecutionEnvironment上调用enablecheckpoint(n),其中n是检查点间隔(以毫秒为单位)。
检查点的其他参数包括:

通过定义检查点之间的时间比检查点间隔更容易配置应用程序,因为“检查点之间的时间”不容易受到检查点有时花费的时间可能超过平均时间这一事实的影响(例如,如果目标存储系统临时变慢)。
注意,这个值还意味着并发检查点的数量是1。

val env = StreamExecutionEnvironment.getExecutionEnvironment()

// 每隔1000 ms进行启动一个检查点
env.enableCheckpointing(1000)

// advanced options:

// 设置模式为exactly-once 默认(this is the default)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

// 确保检查点之间有进行500 ms的进度
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

// 检查点必须在一分钟内完成,或者被丢弃
env.getCheckpointConfig.setCheckpointTimeout(60000)

// prevent the tasks from failing if an error happens in their checkpointing, the checkpoint will just be declined.
env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)

// 同一时间只允许进行一个检查点
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

相关的配置选项

可以通过 conf/flink-conf.yaml设置更多参数或默认值(详见配置):

Key 默认值 描述
state.backend (none) 用于存储和检查点状态的状态后端。
state.backend.async true 选项状态后端是否应在可能且可配置的情况下使用异步快照方法。某些状态后端可能不支持异步快照,或者只支持异步快照,并忽略此选项。
state.backend.fs.memory-threshold 1024 状态数据文件的最小值。所有小于此值的状态块都内联存储在根检查点元数据文件中。
state.backend.incremental false 如果有可能的话,选项状态后端应该创建增量检查点,对于增量检查点,只存储与前一个检查点的差异,而不是完整的检查点状态。一些状态后端可能不支持增量检查点,并忽略此选项。
state.backend.local-recovery false 此选项配置此状态后端的本地恢复。默认情况下,本地恢复是不可用的。本地恢复目前只有键状态后端可用。目前,memorystateback不支持本地恢复,并忽略此选项。
state.checkpoints.dir (none) 用于在支持Flink的文件系统中存储检查点的数据文件和元数据的默认目录。存储路径必须可以从所有参与的流程/节点访问(即所有任务管理器和作业管理器)。
state.checkpoints.num-retained 1 保留已完成的检查点的最大数量。
state.savepoints.dir (none) 保存点的默认目录。用于将保存点写入文件系统的状态后端(memorystate后端、fsstate后端、rocksdbstate后端)。
taskmanager.state.local.root-dirs (none) 配置参数定义根目录,用于存储本地恢复的基于文件的状态。本地恢复目前只覆盖键状态后端。目前,memorystateback不支持本地恢复,并忽略此选项

选择状态后端

Flink的(检查点机制)[https://ci.apache.org/projects/flink/flink-docs-release-1.7/internals/stream_checkpointing.html]在计时器和有状态操作符中存储所有状态的一致快照,包括连接器、窗口和用户定义的状态。检查点存储在何处(例如,JobManager内存、文件系统、数据库)取决于配置的状态后端。

默认情况下,状态保存在Taskmanager的内存中,检查点存储在JobManager的内存中。对于大状态的适当持久性,Flink支持在其他状态后端存储和检查点状态的各种方法。状态后端可以通过streamexecutionenvironment.setStateBackend(…)配置。

有关可用的状态后端以及作业范围和集群范围配置的选项的详细信息,请参见状态后端

迭代作业中的状态检查点

Flink目前仅为没有迭代的作业提供处理保证。在迭代作业上启用检查点会导致异常。为了在迭代程序上强制检查点,用户需要在启用检查点时设置一个特殊的标志:env.enableCheckpointing(interval, force = true)

重新启动策略

Flink支持不同的重启策略,这些策略控制在出现故障时如何重新启动作业。有关更多信息,请参见重新启动策略

上一篇 下一篇

猜你喜欢

热点阅读