1. 版本说明

本文档内容基于 flink-1.13.x，其他版本的整理，请查看本人博客的 flink 专栏其他文章。

2. Checkpoint

2.1. 概述

Checkpoint 使 Flink 的状态具有良好的容错性，通过 checkpoint 机制，Flink 可以对作业的状态和计算位置进行恢复。

参考 Checkpointing 查看如何在 Flink 程序中开启和配置 checkpoint。

2.2. Checkpoint存储

当开启 checkpointing 时，管理的状态会被持久化以保证在任务失败时进行一致性恢复，checkpointing 期间的状态持久化位置取决于选择的 checkpoint 存储。

2.3. 可用的Checkpoint存储选项

flink 捆绑了这些开箱即用的 checkpoint 存储类型：

• JobManagerCheckpointStorage
• FileSystemCheckpointStorage

如果配置了 checkpoint 目录，则会使用 FileSystemCheckpointStorage，否则系统将会使用 JobManagerCheckpointStorage。

2.3.1. JobManagerCheckpointStorage

JobManagerCheckpointStorage 会将 checkpoint 快照存储到 JobManager 的内存中。

可以配置在 checkpoint 使用的内存超过了指定大小时，使 checkpoint 失败，以避免 JobManager 抛出 OutOfMemoryError，为了设置该特性，用户可以实例化一个 JobManagerCheckpointStorage，并指定最大 size：

new JobManagerCheckpointStorage(MAX_MEM_STATE_SIZE);

JobManagerCheckpointStorage 的限制：

• 每个状态的大小默认限制为 5MB，该值可以通过 JobManagerCheckpointStorage 的构造器来设置。
• 和配置的最大状态大小无关，状态不能超过 Akka 框架的大小（查看 Configuration）
• 聚合状态必须适合 JobManager 内存。

JobManagerCheckpointStorage 建议用于：

• 本地开发和调试。
• job 使用很小的状态，比如只包含实时处理数据的函数（Map、FlatMap、Filter、…）， Kafka 消费者只需要很小的状态。

2.3.2. FileSystemCheckpointStorage

FileSystemCheckpointStorage 需要配置一个文件系统 URL（type、address、path），比如：“hdfs://namenode:40010/flink/checkpoints” 或 “file:///data/flink/checkpoints”。

在 checkpointing 时，他会将状态该快照写入到配置的文件系统和路径的文件中，少量元数据将会被存储到 JobManager 的内存中，如果是高可用模式，则存储到元数据 checkpoint 中。

如果指定了 checkpoint 目录，FileSystemCheckpointStorage 将会被用于持久化 checkpoint 快照。

FileSystemCheckpointStorage 建议用于：

• 所有高可用设置

同时也建议设置 managed memory 为 0，该设置会确保在 JVM 上分配最大数量的内存给用户代码。

2.4. 保留Checkpoint

Checkpoint 在默认情况下仅用于恢复失败的作业，并不会保留，当程序取消时 checkpoint 就会被删除。当然，你可以通过配置来保留 checkpoint，这些被保留的 checkpoint 在作业失败或取消时不会被清除。如此一来，你就可以使用该 checkpoint 来恢复失败的作业了。

CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

ExternalizedCheckpointCleanup 配置项定义了当作业取消时，对作业 checkpoint 的操作：

• ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION：当作业取消时，保留作业的 checkpoint。注意，在这种情况下，需要手动清除该作业保留的 checkpoint。
• ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：当作业取消时，删除作业的 checkpoint。仅当作业失败时，作业的 checkpoint 才会被保留。

2.4.1. 目录结构

与 savepoints 相似，checkpoint 由元数据文件和数据文件（与状态后端相关）组成。可通过配置文件中的 “state.checkpoints.dir” 配置项来指定元数据文件和数据文件的存储路径，另外也可以在代码中针对单个作业指定该配置项。

当前的 checkpoint 目录结构如下所示:

/user-defined-checkpoint-dir
    /{job-id}
        |
        + --shared/
        + --taskowned/
        + --chk-1/
        + --chk-2/
        + --chk-3/
        ...        

其中 SHARED 目录保存了会被多个 checkpoint 引用的文件，TASKOWNED 保存了不会被 JobManager 删除的文件，EXCLUSIVE 则保存那些仅被单个 checkpoint 引用的文件。

注意: Checkpoint 目录不是公共 API 的一部分，因此可能会在未来的发版中进行改变。

2.4.1.1. 通过配置文件设置全局配置

state.checkpoints.dir: hdfs:///checkpoints/

2.4.1.2. 对单个job配置checkpoint

env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data/"));

2.4.1.3. 配置checkpoint存储实例

另外，可以通过指定预期的 checkpoint 存储实例来设置 checkpoint 存储，该实例允许设置底层配置，比如写入缓存大小。

env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs:///checkpoints-data/", FILE_SIZE_THESHOLD));

2.4.2. Checkpoint与 Savepoint区别

Checkpoint 与 savepoints 有一些区别，体现在 checkpoint ：

• 使用状态后端特定的数据格式，可能以增量方式存储。
• 不支持 Flink 的特定功能，比如扩缩容。

2.4.3. 从保留的checkpoint中恢复状态

与 savepoint 一样，作业可以从 checkpoint 的元数据文件恢复运行（savepoint恢复指南）。注意，如果元数据文件中信息不充分，则 jobmanager 还需要使用相关的数据文件来恢复作业(参考目录结构)。

$ bin/flink run -s :checkpointMetaDataPath [:runArgs]

2.4.4. 非对齐checkpoints

从 Flink 1.11 开始，可以使用非对齐 checkpoint了。非对齐 checkpoints 包含还未处理的数据（比如存储在缓存中的数据），会将其作为 checkpoint 状态的一部分，该行为可以让 checkpoint barriers 追上这些缓存。如此一来，checkpint 的持续时间就不会受到当前吞吐量的影响了，因为 checkpoint barriers 不再有效的嵌入到数据流中了。

当由于过高的反压导致 checkpoint 持续时间很长时，就可以使用非对齐 checkpoit 了，之后，checkpoint 的时间就不会受到端到端延迟的影响了。注意，非对齐 checkpoint 会增加状态后端的 I/O，因此，在 I/O 是 checkpoint 期间使用状态后端的瓶颈时，就不应该使用非对齐 checkpoint 了。

注意，非对齐 checkpoint 目前是新特性，并且有以下限制：

• Flink 目前不支持并行非对齐 checkpoint，但是 checkpoint 是可预测的，并且时间一般较短，因此可能也不需要并行 checkpoint。然而，savepoint 不能和非对齐 checkpoint 同事进行，因此 savepoint 会执行较长时间。
• 非对齐 checkpoint 会在恢复期间对水印进行隐式保证。

目前，Flink 会在恢复的第一步就生成水印，而不是将算子中最新水印保存起来，以减少状态大小。使用非对齐 checkpoint 进行恢复时，Flink 会在恢复完还未处理的数据后生成水印。如果你的 pipeline 中使用需要接收每个数据最新水印的操作的话，这将会产出和非对齐 checkpoint 不同的结果。如果你的操作取决于永远可用的最新水印，则解决方案为将水印存储到算子状态中。为了支持减少大小，水印应该存储到合并状态的每个 key-group 中。我们很可能会将此方法作为通用解决方案使用（在 Flink 1.11.0 中未成功）。

启用非对齐 checkpoint 后，你也可以通过 CheckpointConfig.setAlignmentTimeout(Duration) 或 execution.checkpointing.alignment-timeout 在配置文件中指定对齐超时时间。指定超时时间之后，每个 checkpoint 刚开始任然是对齐的 checkpoint，但是如果某些子任务的对齐时间超过了该超时时间，就会变成非对齐 checkpoint。

3. Savepoint

3.1. 概念以及与checkpoint的不同

Savepoint 是依据 Flink checkpointing 机制所创建的流作业执行状态的一致镜像。你可以使用 Savepoint 进行 Flink 作业的停止与重启、克隆或更新。 Savepoint 由两部分组成：稳定存储（比如 HDFS，S3，…) 上包含二进制文件的目录（通常很大）和元数据文件（相对较小）。稳定存储上的文件表示作业执行状态的数据镜像。Savepoint 的元数据文件主要包含以相对路径的形式指向稳定存储上所有文件的指针。

注意: 为了允许程序和 Flink 版本之间的升级，请务必查看以下有关分配算子 ID 的部分 。

从概念上讲， Flink 的 Savepoint 与 Checkpoint 的不同之处类似于传统数据库中的备份与恢复日志之间的差异。 Checkpoint 的主要目的是为意外失败的作业提供恢复机制。 Checkpoint 的生命周期由 Flink 管理，即 Flink 创建，管理和删除 Checkpoint，无需用户交互。作为一种恢复和定期触发的方法，Checkpoint 实现有两个设计目标：

1. 轻量级创建
2. 尽可能快地恢复。

可能会利用某些特定的属性来达到这个，例如， 工作代码在执行尝试之间不会改变。 在用户终止作业后，通常会删除 Checkpoint（除非明确配置为保留的 Checkpoint）。

与此相反、Savepoint 由用户创建，拥有和删除。 他们的用例是计划的，手动备份和恢复。 例如，升级 Flink 版本，调整用户逻辑，改变并行度，以及进行红蓝部署等。 当然，Savepoint 必须在作业停止后继续存在。 从概念上讲，Savepoint 的生成和恢复成本可能更高一些，Savepoint 更多地关注可移植性以及对作业更改的支持。

Flink 对所有的状态后端使用了统一的二进制格式，这意味着你可以使用过一种状态后端触发 savepoint，然后使用另一种状态后端来恢复作业。

在 1.13 版本之前，状态后端是没有统一的格式的，因此，如果你想切换作业使用的状态后端，你应该先升级你的 Flink 版本，然后使用新版本触发一个 savepoint，只有做完这些之后，你才能使用另一种状态后端来恢复作业。

3.2. 分配算子ID

强烈建议你按照本节所述调整你的程序，以便将来能够升级你的程序。主要是通过 uid(String) 方法手动指定算子 ID ，这些 ID 将用于恢复每个算子的状态。

DataStream<String> stream = env.
State backends did not start producing a common format until version 1.13
  .addSource(new StatefulSource())
  .uid("source-id") // source 算子的 ID
  .shuffle()
  // 有状态的 mapper 算子的 ID
  .map(new StatefulMapper())
  .uid("mapper-id") // 指定 mapper 算子的 ID
  // 无状态的 printing sink
  .print(); // 自动生成 ID

如果不手动指定 ID ，则会自动生成 ID 。只要这些 ID 不变，就可以从 Savepoint 自动恢复。生成的 ID 取决于程序的结构，并且对程序更改很敏感。因此，强烈建议手动分配这些 ID 。

3.2.1. Savepoint状态

你可以将 Savepoint 想象为对每个有状态的算子都保存一个“算子 ID ->状态”的映射:

Operator ID | State
------------+------------------------
source-id   | State of StatefulSource
mapper-id   | State of StatefulMapper

在上面的示例中，print sink 是无状态的，因此不是 Savepoint 状态的一部分。默认情况下，我们会尝试将 Savepoint 的每个 entry 都映射到新程序。

3.3. 算子

你可以使用命令行客户端来触发 Savepoint，触发 Savepoint 并取消作业，从 Savepoint 恢复，以及删除 Savepoint。

从 Flink 1.2.0 开始，还可以使用 webui 从 Savepoint 恢复。

3.3.1. 触发Savepoint

在触发 Savepoint 时，会创建一个新的 Savepoint 目录，用来存储数据和元数据。可以通过配置默认目标目录或使用命令来指定目标目录(参见:targetDirectory参数）来控制该目录的位置。

**注意:**目标目录必须是 JobManager(s) 和 TaskManager(s) 都可以访问的位置，例如分布式文件系统（或对象存储系统）上的位置。

以 FsStateBackend 或 RocksDBStateBackend 为例：

# Savepoint 目标目录
/savepoint/

# Savepoint 目录
/savepoint/savepoint-:shortjobid-:savepointid/

# Savepoint 文件包含 Checkpoint元数据
/savepoint/savepoint-:shortjobid-:savepointid/_metadata

# Savepoint 状态
/savepoint/savepoint-:shortjobid-:savepointid/...

从 1.11.0 开始，你可以通过移动（拷贝）savepoint 目录到任何地方，然后再进行恢复。

如下两种情况不支持 savepoint 目录的移动：

1. 启用了 entropy injection：在这种情况下，savepoint 目录就不会包含所有的数据文件了，因为注入的路径会分散在各个路径中。由于缺乏一个共同的根目录，所以 savepoint 会包含绝对路径，从而导致无法支持 savepoint 目录的迁移。
2. 作业包含 task-owned state（比如 GenericWriteAhreadLog sink）。

和 savepoint 不同，checkpoint 不支持随意移动文件，因为 checkpoint 可能包含一些文件的绝对路径。

如果你使用了 MemoryStateBackend ，则 metadata 和 savepoint 的数据都会保存在 _metadata 文件中，因此不要因为看到目录下没有数据文件而感到困惑。

注意: 不建议移动或删除正在运行作业的最后一个 Savepoint ，这可能会干扰故障恢复。因此，Savepoint 对精确一次的接收器有副作用，为了确保精确一次的语义，如果在最后一个 Savepoint 之后没有 Checkpoint ，则会使用 Savepoint 进行恢复。

3.3.1.1. 触发 Savepoint

$ bin/flink savepoint :jobId [:targetDirectory]

这将触发 ID 为 :jobId 的作业的 Savepoint，并返回创建的 Savepoint 路径。 你需要此路径来还原和删除 Savepoint 。

3.3.1.2. 使用 YARN 触发 Savepoint

$ bin/flink savepoint :jobId [:targetDirectory] -yid :yarnAppId

这将触发 ID 为 :jobId 和 YARN 应用程序 ID 为 :yarnAppId 的作业的 Savepoint，并返回创建的 Savepoint 的路径。

3.3.1.3. 使用 Savepoint 停止作业

$ bin/flink stop -s [:targetDirectory] :jobId

这将自动触发 ID 为 :jobid 的作业的 Savepoint，并停止该作业。此外，你也可以指定一个目标文件系统目录来存储 Savepoint 。该目录需要能被 JobManager(s) 和 TaskManager(s) 访问。

3.3.2. 从 Savepoint 恢复

$ bin/flink run -s :savepointPath [:runArgs]

这将提交作业并指定 Savepoint 路径。你可以指定 Savepoint 目录或 _metadata 文件的路径。

3.3.2.1. 跳过无法映射的状态恢复

默认情况下，恢复操作会尝试将 Savepoint 的所有状态映射回你要还原的程序。如果新程序删除了算子，则可以通过 --allowNonRestoredState（短命令：-n）选项跳过无法映射到新程序的状态：

$ bin/flink run -s :savepointPath -n [:runArgs]

3.3.3. 删除 Savepoint

$ bin/flink savepoint -d :savepointPath

这将删除存储在 :savepointPath 路径中的 Savepoint。

请注意，还可以通过常规文件系统操作手动删除 Savepoint ，而不会影响其他 Savepoint 或 Checkpoint（请记住，每个 Savepoint 都是自包含的）。在 Flink 1.2 之前，执行上面的 Savepoint 命令是一个很啰嗦的任务。

3.3.4. 配置

你可以通过 state.savepoints.dir 配置 savepoint 的默认目录，触发 savepoint 时，将使用此目录来存储 savepoint。可以使用触发命令指定自定义目录来覆盖缺省值（请参阅:targetDirectory参数）。

flink-conf.yaml

# 默认的 savepoint 目录
state.savepoints.dir: hdfs:///flink/savepoints

Java

env.setDefaultSavepointDir("hdfs:///flink/savepoints");

Scala

env.setDefaultSavepointDir("hdfs:///flink/savepoints")

如果既未配置缺省值也未指定自定义目目录，则触发 Savepoint 的操作将会失败。

**注意:**目标目录必须是 JobManager(s) 和 TaskManager(s) 可访问的位置，例如分布式文件系统上的位置。

3.4. F.A.Q

1. 需要为所有算子分配 ID 吗

根据经验，是的。严格来说，只需要通过 uid 方法给有状态算子分配 ID 就足够了。Savepoint 仅包含这些有状态算子的状态，无状态算子不是 Savepoint 的一部分。

在实践中，建议给所有算子分配 ID，因为 Flink 的一些内置算子（如 Window 算子）也是有状态的，而内置算子是否有状态并不很明显。如果你完全确定某个算子是无状态的，则可以不调用 uid 方法。

2. 如果我在作业中添加一个需要状态的新算子，会发生什么

当你向作业添加新算子时，它将在没有任何状态的情况下进行初始化。 Savepoint 包含了每个有状态算子的状态，无状态算子根本不是 Savepoint 的一部分。新算子相当于无状态算子。

3. 如果从作业中删除有状态的算子会发生什么

默认情况下，从 Savepoint 恢复时会尝试将所有状态分配给新作业。如果有状态算子被删除，则无法从 Savepoint 恢复。

你可以通过使用 run 命令设置 --allowNonRestoredState (简称：-n )来允许删除有状态算子:

$ bin/flink run -s :savepointPath -n [:runArgs]

4. 如果我在作业中重新排序了有状态算子，会发生什么

如果给这些算子分配了 ID，它们将会像往常一样恢复。

如果没有分配 ID ，则有状态算子自动生成的 ID 很可能在重新排序后发生更改。这将导致你无法从之前的 Savepoint 恢复。

5. 如果我添加、删除或重新排序作业中没有状态的算子，会发生什么

如果将 ID 分配给有状态算子，则无状态算子不会影响 Savepoint 恢复。

如果没有分配 ID ，则有状态算子自动生成的 ID 很可能在重新排序后发生更改。这将导致你无法从以前的Savepoint 恢复。

6. 当我在恢复时改变程序的并行度时会发生什么

如果 Savepoint 是用 Flink >= 1.2.0 触发的，并且没有使用像 Checkpointed 这样的不推荐的状态 API，那么你可以简单地从 Savepoint 恢复程序并指定新的并行度。

如果你正在从 Flink < 1.2.0 触发的 Savepoint 恢复，或者使用现在已经废弃的 api，那么你首先必须将作业和 Savepoint 迁移到 Flink >= 1.2.0，然后才能更改并行度。参见升级作业和Flink版本指南。

7. 我可以将 savepoint 文件移动到稳定存储上吗

这个问题的答案目前是“是”，从 Flink 1.11.0 版本开始，savepoint 是自包含的，你可以按需迁移 savepoint 文件后进行恢复。

4. 状态后端

用 Data Stream API 编写的程序通常会以各种形式保存状态：

• 在 Window 触发之前要么收集元素、要么执行聚合
• 转换函数可以使用 key/value 格式的状态接口来存储状态
• 转换函数可以实现 CheckpointedFunction 接口，使函数的本地变量具有容错能力

另请参阅 Streaming API 指南中的 状态部分 。

在启动 CheckPoint 机制时，状态会随着 CheckPoint 而持久化，以防止数据丢失、保障恢复时的一致性。状态内部的存储格式、状态在 CheckPoint 时如何持久化以及持久化在哪里均取决于选择的 State Backend。

4.1. 可用的状态后端

Flink 内置了以下这些开箱即用的 state backends ：

• HashMapStateBackend
• EmbeddedRocksDBStateBackend

如果不设置，则默认使用 HashMapStateBackend。

4.1.1. HashMapStateBackend

在 HashMapStateBackend 内部，数据以 Java 对象的形式存储在堆中。 Key/value 形式的状态和窗口算子会持有一个 hash table，其中存储着状态值、触发器。

HashMapStateBackend 的适用场景：

• 有较大 state，较长 window 和较大 key/value 状态的 Job。
• 所有的高可用场景。

建议同时将 managed memory 设为 0，以保证最大限度的将内存分配给 JVM 上的用户代码。

4.1.2. EmbeddedRocksDBStateBackend

EmbeddedRocksDBStateBackend 会将正在运行中的状态数据保存在 RocksDB 数据库中，RocksDB 数据库默认将数据存储在 TaskManager 的数据目录。不同于 HashMapStateBackend 中的 java 对象，数据以序列化字节数组的方式存储，这种方式由序列化器决定，因此 key 之间的比较是以字节序的形式进行而不是使用 Java 的 hashCode 或 equals() 方法。

EmbeddedRocksDBStateBackend 会使用异步的方式生成 snapshots。

EmbeddedRocksDBStateBackend 的局限：

• 由于 RocksDB 的 JNI API 构建在 byte[] 数据结构之上, 所以每个 key 和 value 最大支持 2^31 个字节。 RocksDB 合并操作的状态（例如：ListState）累积数据量大小可以超过 2^31 字节，但是会在下一次获取数据时失败。这是当前 RocksDB JNI 的限制。

EmbeddedRocksDBStateBackend 的适用场景：

• 状态非常大、窗口非常长、key/value 状态非常大的 Job。
• 所有高可用的场景。

注意，你可以保留的状态大小仅受磁盘空间的限制。与状态存储在内存中的 HashMapStateBackend 相比，EmbeddedRocksDBStateBackend 允许存储非常大的状态。 然而，这也意味着使用 EmbeddedRocksDBStateBackend 将会降低应用程序的最大吞吐量。所有的读写都必须进行序列化、反序列化操作，这个要比基于堆内存的状态后端的效率低很多。

请同时参考 Task Executor 内存配置 中关于 EmbeddedRocksDBStateBackend 的建议。

EmbeddedRocksDBStateBackend 是目前唯一支持增量 CheckPoint 的状态后端(见 这里)。

flink 允许获取一些 RocksDB 的本地指标(metrics)，默认是关闭的。你能在 这里 找到关于 RocksDB 本地指标的文档。

每个 slot 中的 RocksDB 实例的内存大小是有限制的，详情请见 这里。

4.2. 选择合适的状态后端

在选择 HashMapStateBackend 和 RocksDB 的时候，其实就是在性能与可扩展性之间的权衡。HashMapStateBackend 是非常快的，因为每个状态的读取和算子对于对象的更新都是在 Java 的 heap 上进行的，但状态的大小受限于集群中可用的内存。另一方面，RocksDB 可以根据可用的磁盘空间扩展，并且只有它支持增量快照。然而，每个状态的读取和更新都需要序列化和反序列化，而且在 disk 上进行读操作的性能可能要比基于内存的状态后端慢一个数量级。

在 Flink 1.13 版本中我们统一了 savepoints 的二进制格式。这意味着你可以生成 savepoint 并且之后使用另一种状态后端读取它。从 1.13 版本开始，所有的状态后端都会生成一种普遍适用的格式。因此，如果想切换状态后端的话，最好先升级你的 Flink 版本，在新版本中生成 savepoint，在这之后你才可以使用另一个不同的状态后端来读取并恢复它。

4.2.1. 设置状态后端

如果没有明确指定，将使用 jobmanager 做为默认的状态后端。你能在 flink-conf.yaml 中为所有 Job 设置其他默认的状态后端。每一个 Job 的状态后端配置都会覆盖默认的状态后端配置，如下所示。

4.2.1.1. 设置每个 Job 的状态后端

StreamExecutionEnvironment 可以对每个 Job 的状态后端进行设置，如下所示：

Java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());

Scala

val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.setStateBackend(new HashMapStateBackend())

如果你想在 IDE 中使用 EmbeddedRocksDBStateBackend，或者需要在作业中通过编程方式进行动态配置，必须将以下依赖添加到 Flink 项目中。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
    <version>1.13.6</version>
    <scope>provided</scope>
</dependency>

注意: 由于 RocksDB 是 Flink 默认分发包的一部分，所以如果你没在代码中使用 RocksDB，则不需要添加此依赖。而且可以在 flink-conf.yaml 文件中通过 state.backend 配置状态后端，以及更多的 checkpointing 和 RocksDB 特定的 参数。

4.2.1.2. 设置默认的全局状态后端

在 flink-conf.yaml 可以通过键 state.backend 设置默认的状态后端。

可选值包括 jobmanager (HashMapStateBackend), rocksdb (EmbeddedRocksDBStateBackend)， 或使用实现了状态后端工厂 StateBackendFactory 的类的全限定类名，例如： EmbeddedRocksDBStateBackend 对应为 org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackendFactory。

state.checkpoints.dir 选项指定所有状态后端写 CheckPoint 数据和元数据文件的目录。你能在 这里 可以找到关于 CheckPoint 目录结构的详细信息。

配置文件的部分示例如下所示：

# 用于存储 operator state 快照的 State Backend
state.backend: filesystem

# 存储快照的目录
state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

4.3. RocksDB状态后端进阶

该小节描述 RocksDB 状态后端的更多细节。

4.3.1. 增量快照

RocksDB 支持增量快照。不同于产生一个包含所有数据的全量备份，增量快照中只包含自上一次快照完成之后被修改的记录，因此可以显著减少快照完成的耗时。

增量快照是基于（通常是多个）前序快照构建的。由于 RocksDB 内部的压缩机制会对 sst 文件进行合并，所以 Flink 的增量快照也会定期压缩，因此增量 checkpoint 历史不会一直增长，旧快照包含的文件也会逐渐过期并被自动清理。

和基于全量快照的恢复时间相比，如果网络带宽成为瓶颈，则基于增量快照恢复可能会消耗更多时间，因为增量快照包含的 sst 文件之间可能存在数据重叠导致需要下载的数据量变大；而当 CPU 或 IO 是瓶颈的时候，基于增量快照的恢复会更快，因为从增量快照恢复并不需要解析 Flink 的统一快照格式来重建本地的 RocksDB 数据表，而是直接基于 sst 文件进行加载。

虽然状态很大时我们推荐使用增量快照，但这并不是默认的快照机制，需要通过下述配置手动开启该功能：

• 在 flink-conf.yaml 中设置：state.backend.incremental: true 或者
• 在代码中按照右侧配置方式以覆盖默认配置：EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);

需要注意的是，一旦启用了增量快照，网页上展示的 Checkpointed Data Size 就只代表增量上传的数据量了，而不是一次快照的完整数据量。

4.3.2. 内存管理

Flink 致力于控制整个进程的内存消耗，以确保 Flink TaskManager 有良好的内存使用，从而既不会在容器（Docker/Kubernetes, Yarn等）环境中由于内存超用被杀掉，也不会因为内存利用率过低导致不必要的数据落盘或是缓存命中率下降，从而致使性能下降。

为了达到上述目标，Flink 默认将 RocksDB 的可用内存配置为 TaskManager 的单槽（per-slot）托管内存量。这将为大多数应用程序提供良好的开箱即用的体验，即大多数应用程序不需要调整 RocksDB 配置，简单的增加 Flink 的托管内存即可改善内存相关性能问题。

当然，也可以选择不使用 Flink 自带的内存管理，而是手动为 RocksDB 的每个列族（ColumnFamily）分配内存（每个算子的每个状态都对应一个列族）。这为专业用户提供了对 RocksDB 进行更细粒度控制的途径，但同时也意味着用户需要自行保证总内存消耗不会超过（尤其是容器）环境的限制。请参阅 大状态调优 了解有关大状态数据性能调优的一些指导原则。

RocksDB 使用托管内存

这个功能是默认打开的，并且可以通过 state.backend.rocksdb.memory.managed 配置项控制。

Flink 并不直接控制 RocksDB 的本地内存分配，而是通过配置 RocksDB 来确保其使用的内存正好与 Flink 的预算托管内存相同。这是在任务槽（per-slot）级别上完成的（托管内存以任务槽为粒度计算）。

为了设置 RocksDB 实例的总内存使用量，Flink 对同一个任务槽上的所有 RocksDB 实例使用共享缓存以及写入缓存管理器。 共享缓存将对 RocksDB 中内存消耗的三个主要来源（块缓存、索引和布隆过滤器、MemTables）设置上限。

Flink 还提供了两个参数来控制写路径（MemTable）和读路径（索引及过滤器，读缓存）之间的内存分配。当你遇到 RocksDB 由于缺少写缓冲内存（频繁刷新）或读缓存未命中而导致性能不佳时，可以使用这些参数调整读写间的内存分配。

• state.backend.rocksdb.memory.write-buffer-ratio，默认值 0.5，即 50% 的给定内存会分配给写缓冲区使用。
• state.backend.rocksdb.memory.high-prio-pool-ratio，默认值 0.1，即 10% 的快缓存内存会优先分配给索引及过滤器。我们强烈建议不要将此值设置为零，以防止索引和过滤器被频繁踢出缓存而导致性能问题。此外，我们默认将 L0 级的过滤器和索引固定到缓存中以提高性能，更多详细信息请参阅 RocksDB 文档。

注意：上述机制开启时将覆盖用户在 PredefinedOptions 和 RocksDBOptionsFactory 中对快缓存和写缓存进行的配置。

注意：仅面向专业用户：若要手动控制内存，可以将 state.backend.rocksdb.memory.managed 设置为 false，并通过 ColumnFamilyOptions 配置 RocksDB。或者可以复用上述 缓存/写入缓存管理器 机制，将内存大小设置为与 Flink 的托管内存大小无关的固定大小（通过 state.backend.rocksdb.memory.fixed-per-slot 选项）。 注意在这两种情况下，用户都需要确保在 JVM 之外有足够的内存可供 RocksDB 使用。

4.3.3. 定时器（内存 vs.RocksDB）

定时器（Timer）用于执行定时操作（基于事件时间或处理时间），例如触发窗口或回调 ProcessFunction。

当选择 RocksDB 作为状态后端时，默认情况下定时器也存储在 RocksDB 中。这是一种健壮且可扩展的方式，允许应用程序使用很多个定时器。另一方面，在 RocksDB 中维护定时器会有一定的成本，因此 Flink 也提供了将定时器存储在 JVM 堆上而使用 RocksDB 存储其他状态的选项。当定时器数量较少时，基于堆的定时器会有更好的性能。

可以通过将 state.backend.rocksdb.timer-service.factory 配置项设置为 heap 而不是默认的 rocksdb 来将定时器存储在堆上。

注意：在 RocksDB 状态后端中使用基于堆的定时器的组合当前不支持定时器状态的异步快照。其他状态（如 keyed state）可以进行异步快照。

当使用基于 java heap 的定时器的 RocksDB 状态后端时，如果程序中有算子写入原生 keyed 状态的话，则预期的 checkpointing 和触发 savepoint 的操作会发生异常而失败，这是为编写自定义流算子的高级用户提供的功能。

4.3.4. 开启 RocksDB 原生监控指标

您可以选择使用 Flink 的监控指标系统来汇报 RocksDB 的原生指标，并且可以选择性的指定特定指标进行汇报。请参阅 配置文档 了解更多详情。

注意：启用 RocksDB 的原生指标可能会对应用程序的性能产生负面影响。

4.3.5. RocksDB内存高级调优

Flink 对大多数用例都提供了详细的默认 RocksDB 内存管理，下面的机制只是为了专业调优和问题解决使用。

4.3.5.1. 列族（ColumnFamily）级别的预定义选项

使用预定义选项，用户可以在每个 RocksDB 列族上应用一些预定义的配置，例如配置内存使用、线程、压缩设置等。目前每个算子的每个状态都在 RocksDB 中有专门的一个列族存储。

有两种方法可以选择要应用的预定义选项：

• 在配置文件 flink-conf.yaml 中设置 state.backend.rocksdb.predefined-options 配置项。
• 通过程序设置：EmbeddedRocksDBStateBackend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM) 。

该选项的默认值是 DEFAULT ，对应 PredefinedOptions.DEFAULT 。

程序中设置的预定义选项会覆盖 flink-conf.yaml 文件中的配置。

4.3.5.2. 通过 RocksDBOptionsFactory 配置 RocksDB 选项

您也可以通过配置一个 RocksDBOptionsFactory 来手动控制 RocksDB 的配置选项。此机制使您可以对列族的设置进行细粒度控制，例如内存使用、线程、压缩设置等。目前每个算子的每个状态都在 RocksDB 中有专门的一个列族存储。

有两种方法可以将 RocksDBOptionsFactory 传递给 RocksDB 状态后端：

• 在配置文件 flink-conf.yaml 中设置 state.backend.rocksdb.options-factory 配置项。
• 通过程序设置，例如 EmbeddedRocksDBStateBackend.setRocksDBOptions(new MyOptionsFactory()); 。

注意：通过程序设置的 RocksDBOptionsFactory 将覆盖 flink-conf.yaml 配置文件中的设置，且 RocksDBOptionsFactory 设置的优先级高于预定义选项（PredefinedOptions）。

注意：RocksDB是一个本地库，会直接从进程中分配内存，而不是从 JVM 分配内存。分配给 RocksDB 的任何内存都必须被考虑在内，通常需要将这部分内存从 TaskManager 的 JVM 堆中减去。不这样做可能会导致JVM进程由于分配的内存超过申请值而被 YARN/Mesos 等资源管理框架终止。

从 flink-conf.yaml 中读取列族选项

实现了 ConfigurableRocksDBOptionsFactory 接口的 RocksDBOptionsFactory 可以直接从配置文件 flink-conf.yaml 中读取配置。

state.backend.rocksdb.options-factory 的默认配置是 org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory，它默认会将 这里定义 的所有配置项全部加载。 因此您可以简单的通过关闭 RocksDB 使用托管内存的功能并将需要的设置选项加入配置文件来配置底层的列族选项。

下面是自定义 ConfigurableRocksDBOptionsFactory 的一个示例 (开发完成后，需要将实现类全限定名设置到 state.backend.rocksdb.options-factory).

public class MyOptionsFactory implements ConfigurableRocksDBOptionsFactory {
    public static final ConfigOption<Integer> BLOCK_RESTART_INTERVAL = ConfigOptions
            .key("my.custom.rocksdb.block.restart-interval")
            .intType()
            .defaultValue(16)
            .withDescription(
                    " Block restart interval. RocksDB has default block restart interval as 16. ");

    private int blockRestartInterval = BLOCK_RESTART_INTERVAL.defaultValue();

    @Override
    public DBOptions createDBOptions(DBOptions currentOptions,
                                     Collection<AutoCloseable> handlesToClose) {
        return currentOptions
                .setIncreaseParallelism(4)
                .setUseFsync(false);
    }

    @Override
    public ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions,
                                                   Collection<AutoCloseable> handlesToClose) {
        return currentOptions.setTableFormatConfig(
                new BlockBasedTableConfig()
                        .setBlockRestartInterval(blockRestartInterval));
    }

    @Override
    public RocksDBOptionsFactory configure(ReadableConfig configuration) {
        this.blockRestartInterval = configuration.get(BLOCK_RESTART_INTERVAL);
        return this;
    }
}

4.4. 从旧版本迁移

从 Flink 1.13 版本开始，社区改进了状态后端的公开类，进而帮助用户更好理解本地状态存储和 checkpoint 存储的区分。这个变化并不会影响状态后端和 checkpointing 过程的运行时实现和机制，仅仅是为了更好地传达设计意图。用户可以将现有作业迁移到新的 API，同时不会损失原有状态。

4.4.1. MemoryStateBackend

旧版本的 MemoryStateBackend 等价于使用 HashMapStateBackend 和 JobManagerCheckpointStorage。

4.4.1.1. flink-conf.yaml 配置

state.backend: hashmap

# 可选，如果没有指定 checkpoint 目录，则 flink 会默认使用 JobManagerCheckpointStorage
state.checkpoint-storage: jobmanager

Java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(new JobManagerStateBackend());

Scala

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStateBackend(new HashMapStateBackend)
env.getCheckpointConfig().setCheckpointStorage(new JobManagerStateBackend)

4.4.2. FsStateBackend

旧版本的 FsStateBackend 等价于使用 HashMapStateBackend 和 FileSystemCheckpointStorage。

4.4.2.1. flink-conf.yaml 配置

state.backend: hashmap
state.checkpoints.dir: file:///checkpoint-dir/

# 可选，如果没有指定 checkpoint 目录，则 flink 会默认使用 FileSystemCheckpointStorage
state.checkpoint-storage: filesystem

4.4.2.2. 代码配置

Java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");

// FsStateBackend 的高级配置，比如写入缓存大小可以通过手动初始化 FileSystemCheckpointStorage 对象来设置。
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"));

Scala

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStateBackend(new HashMapStateBackend)
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir")

// FsStateBackend 的高级配置，比如写入缓存大小可以通过手动初始化 FileSystemCheckpointStorage 对象来设置。
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"))

4.4.3. RocksDBStateBackend

旧版本的 RocksDBStateBackend 等价于使用 EmbeddedRocksDBStateBackend 和 FileSystemCheckpointStorage.

4.4.3.1. flink-conf.yaml 配置

state.backend: rocksdb
state.checkpoints.dir: file:///checkpoint-dir/

# 可选，如果没有指定 checkpoint 目录，则 flink 会默认使用 FileSystemCheckpointStorage
state.checkpoint-storage: filesystem

4.4.3.2. 代码配置

Java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");

// 如果你想通过在 RocksDBStateBackend 构造器中指定 FsStateBackend 来配置 checkpoint ，比如写入缓存大小，则可以自己初始化一个 FileSystemCheckpointStorage 对象来完成该目标。
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"));

Scala

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStateBackend(new EmbeddedRocksDBStateBackend)
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir")
    
// 如果你想通过在 RocksDBStateBackend 构造器中指定 FsStateBackend 来配置 checkpoint ，比如写入缓存大小，则可以自己初始化一个 FileSystemCheckpointStorage 对象来完成该目标。
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"))

5. 大状态与Checkpoint调优

该张姐描述如何配置和优化使用大状态的程序。

5.1. Overview

为了让 Flink 程序能够大规模可靠的运行，必须满足以下两个条件：

• 程序需要能够可靠的进行 checkpoint。
• 在任务失败后，任务的资源能够满足追上输入的流数据。

第一个条件讨论如果在大规模任务中稳定的进行 checkpoint，第二个条件是有关规划使用多少资源的最佳实践。

5.2. 监控状态和Checkpoints

通过查看 UI 界面是监控 checkpoint 行为最简单的方式，checkpoint 监控 文档描述了如何访问可用 checkpoint 指标。

通过任务级别的指标和web 接口获取到的这两个数字对 checkpoint 调优是非常有用的：

• 从触发 checkpoint 到算子接收到他们的第一个 checkpoint barrier 经常会花费非常多的时间，这意味着 checkpoint barrier 从 source 到算子需要很长的时间，这表明系统正在恒定的反压下运行。
• barrrer 对其花费的时间为接收到第一个和下一个 checkpoint barrier 所花费的时间，精确一次的非对齐 checkpoint 和至少一次的 checkpoint 子任务会处理上游子任务所有的数据而不会发生中断。但是，精确一次的非对齐 checkpoint 中已经接受到 checkpoint barrier 的通道会被阻塞，并且不接受新的数据，直到剩下的通道追赶上，并且接收到他们的 checkpoint barrier（非对齐时间）。

理想情况下这两个值应该很小，较大的值意味着出现了一些反压，从而导致 checkpoint barrier 在作业图中移动的速度很慢，这个现象可以通过处理数据端到端的延迟观察到。注意，在发生瞬间反压、数据倾斜或网络问题时，这些数字会瞬间变大。

非对齐 checkpoints 可以加速 checkpoint barrier 的传播速度，但需要注意，该特性并不能解决导致反压的根本问题，而且端到端的数据延迟依然会很高。

5.3. Checkpointing调优

checkpoint 由程序配置的常规时间间隔来触发，当 checkpoint 花费很长的时间来完成时，则下个 checkpoint 在当前正被处理的 checkpoint 完成前不会被触发。默认情况下，下个 checkpoint 会在正在进行的 checkpoint 完成后马上触发。

如果做完 checkpoint 需要的时间经常超过基础间隔时间，比如状态大小超过预期，或临时性的 checkpoint 存储较慢，则系统会不断的进行 checkpoint，新的 checkpoint 会在正在进行的 checkpoint 完成后马上开始。这意味着太多的资源都花费在了 checkpoint 上，而算子只得到了很少的资源。该行为会对使用同步 checkpoint 状态的流式程序造成较小的影响，但仍然可能会对整个应用性能产生很大的影响。

为了阻止这个情况，程序可以定义两次 checkpoint 之间的最小间隔：

StreamExecutionEnvironment.getCheckpointConfig().setMinPauseBetweenCheckpoints(milliseconds)

最后一次 checkpoint 和下一个新开始的 checkpoint 之间必须经过该最小的时间间隔，下图展示该值如何影响 checkpoint。

注意：程序可以通过 CheckpointConfig 来配置可以同时执行多少个 checkpoint。对于使用了大状态的 flink 程序，这会导致 checkpoint 花费很多资源。当手动触发 savepoint 时，程序有可能也在执行 checkpoint。

5.4. RocksDB调优

很多大状态 flink 流式程序都会使用 RocksDB 为状态后端，该后端可以很好的扩展主内存之外的内存，并且可靠的存储 keyed 状态。

可以通过配置来改变 RocksDB 的执行，该章节为使用 RocksDB 状态后端来优化 job 的最佳实践。

5.4.1. 增量Checkpoints

如果想要减少 checkpoint 花费的时间的话，使用增量 checkpoint 应该该是第一选择。相比于全量 checkpoint，增量 checkpoint 可以明显减少 checkpoint 的时间，因为增量 checkpoint 只会记录与上次完成的 checkpoint 不同的地方，而不是生成一个完整的并且包含状态后端的备份。

查看 RocksDB 的增量 checkpoint 来了解更多信息。

5.4.2. 定时器使用RocksDB或JVM Heap

默认会将定时器存储到 RocksDB，这是个非常好的选择。

在对只含有少量定时器的 job 进行性能优化时，比如没有开窗，没有在 ProcessFunctgion 中使用定时器的 job，将这些定时器放到 heap 内存中是可以提升性能的。请谨慎使用该特性，基于 heap 存储的定时器可能额会增加 checkpoint 时间，而且还不能使用对外内存。

查看 该章节 获取配置基于 heap 定时器的细节。

5.4.3. 调整RocksDB内存

RocksDB 状态后端的性能表现大都取决于可用的内存大小，为了提升性能，增加内存就可以提升很多。

默认情况下，RocksDB 状态后端会使用 Flink 的管理内存预算 RocksDB 缓存，需要设置 state.backend.rocksdb.memory.managed: true。请参考 RocksDB 内存管理 来获取该机制工作的背景知识。

为了调整内存相关性能问题，可以参考以下步骤：

• 提高性能的第一步应该是增加管理内存大小，该操作通常会有很大的改善，而且也不需要调整很多 RocksDB 底层复杂的选项。

  在遇到很大的容器或进程时，RocksDB 会使用很多内存，除非是程序的处理逻辑自己需要很多 JVM heap。默认的管理内存占比 0.4 是一个保守值，在 TaskManager 有好几 GB 的处理内存时，可以增加该占比值。

• 写入缓存的数量取决于你应用程序中的状态数量(pipeline 中所有算子的状态)。每个状态会对应一个列族（columnFamily），并且每个状态都需要自己的写入缓存。因此，有很多状态的程序需要更多的内存。

• 你可以通过设置 state.backend.rocksdb.memory.managed: false 来比较使用管理内存的 RocksDB 和使用单列族内存的性能，这对于基线的测试(假设没有容器内存限制)或与 Flink 早期版本的对比测试会很有用。

  相比于管理内存设置，不使用管理内存意味着 RocksDB 分配的内存与应用程序中的状态数量成正比(内存占用随着应用程序的变化而变化)。根据经验，非管理内存模式(除非使用列族选项)的上限大约为“140MB * 所有任务的状态数量 * slots数量”，定时器也算状态。

• 如果你的程序有很多状态，并且观察到 MemTable 在频繁的刷新（写入侧瓶颈），如果你无法增加总内存，则可以选择增加写入缓存使用的内存比例，通过 state.backend.rocksdb.memory.write-buffer-ratio 来设置。查看 RocksDB 内存管理 来获取更多细节。

• 一个对有很多状态的 job 减少 MemTable 刷新次数的高级选项为通过 RocksDBOptionsFactory 对象调优 RocksDB 的列族选项，比如 arean 块大小、更多的后台刷新线程等。

public class MyOptionsFactory implements ConfigurableRocksDBOptionsFactory {

    @Override
    public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
        // 如果算子有很多状态，可以增加最大后台刷新线程数量，这意味着在一个 DB 示例中有很多列族
        return currentOptions.setMaxBackgroundFlushes(4);
    }

    @Override
    public ColumnFamilyOptions createColumnOptions(
        ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
        // 将默认的 arean 块大小从 8MB 减少为 1MB
        return currentOptions.setArenaBlockSize(1024 * 1024);
    }

    @Override
    public OptionsFactory configure(ReadableConfig configuration) {
        return this;
    }
}

5.5. 容量规划

该节描述如何规划稳定运行 flink job 需要多少资源，容量规划的基础规则如下：

• 正常来说，算子运行应该有足够的资源，以避免出现反压，查看 反压监控 来检查程序运行是否出现了反压。

• 在程序无故障且无反压运行需要的资源基础上提供一些额外的资源，这些资源可以用来追上在程序恢复期间积压的输入数据。至于需要多少额外资源，这取决于算子恢复的所消耗的时间，在恢复期间需要加载到新 TaskManager 的状态大小，以及期望多快恢复完成。
  重要事项：恢复的基础时间点应该是 checkpoint 被激活的时间，因为 checkpoint 也会占用一定的资源，比如网络宽带。

• 暂时性的反压通常是可以接受的，这是在负载峰值、追赶阶段或外部系统(写入sink)出现暂时减速时控制执行流的一个重要部分。

• 一些算子，比如大窗口，会对他们的下游算子造成峰值负载：比如窗口，他的下游算子在窗口构建时并没什么事情可做，但是在窗口触发时，会有很大的负载。下游算子的并行度规划需要考虑有多少窗口会被触发，并且处理峰值能够多快完成。

重要：为了之后能够增加资源，请将 DataStream 程序的 maximum parallelism 设置为一个合理值。最大并行度会决定在修改并恢复程序（通过 savepoint）时可以设置的并行度最大值。

Flink 内部以最大并行度数量的 key group 来跟踪并行状态，即使执行的程序并行度很小，但 flink 仍然可以将最大并行度设置为一个很大的值。

5.6. 压缩

flink 对 checkpoint 和 savepoint 提供了压缩选项，默认关闭。目前使用 snappy compression algorithm (version 1.1.4) 压缩方式，不过我们计划将来支持自定义压缩算法。压缩会在 keyed state 上以 key-group 粒度执行，比如，单独对每个 key-group 进行解压，这对于调整大小是很重要的。

可以通过 ExecutionConfig 来开启压缩：

ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.setUseSnapshotCompression(true);

注：压缩选项对增量快照无影响，因为增量快照使用 RocksDB 的内部格式，该格式通常使用开箱即用的 snappy 压缩。

5.7. Task本地恢复

5.7.1. 动机

在 flink 的 checkpoint 中，每个 task 的状态都会产生一个快照，然后写入到分布式存储。每个 task 都会通过向 JobManager 发送一个包含了状态对应分布式存储位置的句柄来表示状态写入成功，反过来，JobManager 会收集所有 task 发送的句柄，然后将他们捆绑到一个 checkpoint 对象中。

在恢复的时候，JobManager 会打开最新的 checkpoint 对象，然后将句柄返回给对应的 task，之后 task 就可以根据句柄从分布式存储恢复他们的状态了。使用分布式存储来保存状态有两个重要的优势：

1. 存储有容错能力。
2. 分布式存储中的存储的状态可以被所有的节点访问，并且可以很容易的被重新建立，比如并行度调整。

然而，使用远程分布式存储也有一个大的缺点：所有的 task 必须通过网络重远程位置读取它们的状态。在很多情况下，任务恢复是可以将失败的 task 重新调度在上一次运行的相同的 TaskManager 上，当然也有例外，比如机器故障，但我们仍然需要读取远程状态。这会导致即使一些失败只是出现在了单个机器上，大状态恢复也需要很长的时间。

5.7.2. 方式

task 本地状态恢复目标就是为了解决恢复时间过长问题，其主要思想如下：对于每个 checkpoint，每个 task 不仅需要将状态写入分布式存储，同时也需要将状态快照写入到 task 所在机器的本地存储，比如本地磁盘或内存。注意主存储依然是分布式存储，因为本次储存胡无法确保节点失败的持久化，并且不能被其他节点访问。

对于每个可以被重新调度在之前机器来进行恢复的 task，我们就可以从本地存储来恢复状态，本地拷贝可以避免远程状态读取的成本。很多失败并不是节点失败，并且节点失败通常只会影响一个或很少的节点，因此很多 task 可以从他们之前的位置进行恢复，并且找到完整的本地状态，一次你本地恢复更高效，而且可以减少恢复时间。

注意，本地恢复在 checkpoint 的创建和本地存储上会有一些额外的成本，这取决于选择的状态后端和 checkpoint 策略。在大多数情况下，该方案都是通过简单的将分布式存储复制到本地存储。

5.7.3.分布式存储和本地存储的关系

task 本地状态通常会考虑使用本地存储拷贝，checkpoint 状态的成功保障依然是分布式存储，这意味着在 checkpoint 和恢复期间本地存储可能会有问题：

• 对于 checkpoint，分布式存储必须成功才行，本次拷贝失败并不会影响 checkpoint。checkpoint 在分布式拷贝创建失败时会失败，即使本地拷贝被成功创建也不行。
• JobManager 只会成功并管理分布式拷贝，本地拷贝只属于 TaskManager，并且他们的生命周期独立于他们的分布式拷贝。比如，分布式拷贝可能会保留最新的 3 个 checkpoint，但是本地存储只会保留最新的 checkpoint。
• 在恢复时，如果本地拷贝可用，Flink 通常会先尝试从 task 本地状态进行恢复。在从本地拷贝恢复期间发生任何问题，Flink 会马上从远程分布式拷贝来恢复 task。如果分布式拷贝和本地拷贝都失败的话，恢复才会失败。取决于 flink 的配置，在这种情况下，Flink 仍然可以从一个更老的 checkpoint 恢复。
• task 本地拷贝可能只包含完整 task 状态的一部分，比如在写入本地文件时出现异常。在这种情况下，Flink 将会尝试通过本地状态恢复一部分，非本地状态将会从分布式拷贝进行恢复。分布式状态存储必须是完成的，并且是 task 本地状态的合集。
• task 本地状态和分布式状态有不同的格式，并不要求他们完全相同。比如，task 本地状态保存在 heap 内存的对象中，并且没有被存储到任何文件。
• 如果丢失了某个 TaskManager，则它上面的所有 task 本地状态都会丢失。

5.7.4. 配置task本地恢复

task 本地恢复默认是关闭的，可以通过指定 flink 的配置 state.backend.local-recovery 为 CheckpointingOptions.LOCAL_RECOVERY 来激活。该值也可以通过设置为 true 来启用，或设置 false （默认值）来禁用。

注意 非对齐 checkpoints 目前不支持 task 本地恢复。

5.7.5. task本地恢复与不同的状态后端

限制：目前，task 本地恢复只涉及到 keyed 状态后端，keyed 状态通常是状态中最大的一部分。在不久的将来，我们将会涉及到算子状态和定时器。

下面的状态后端支持 task 本地恢复：

• HashMapStateBackend：task 本地恢复支持 keyed 状态，通过拷贝状态到本地文件来实现，会造成额外的写入成本，并且占用本地磁盘空间。在未来，我们可能会提供保存 task 本地状态到内存的实现。
• EmbeddedRocksDBStateBackend：task 本地恢复支持 keyed 状态。对于完成 checkpoint，状态会被拷贝到本地文件，这会造成额外的写入成本并且占用本地磁盘空间。对于增量快照，本地状态基于 RocksDB 的原生 checkpoint 机制。该机制首先会创建分布式拷贝，这意味着在这种情况下，创建本地拷贝并不会产生额外的成本。我们会在上传分布式存储之后，简单的保存原生 checkpoint 目录，而不会删除它。该本地拷贝会通过硬链接分享 RocksDB 工作目录中的活跃文件，因此，对于活跃文件，通过增量快照进行 task 本地恢复，并不会占用额外的空间。使用硬链接意味着 RocksDB 目录必须在同一个物理设备上，并且所有配置本地恢复目录都可以被用来存储本地状态，否则硬链接建立会失败。目前，当 RocksDB 目录被配置到不同的物理设备上时，会阻止使用本地恢复。

5.7.6. 分配保存调度

task 本地恢复会在失败时使用分配保存 task 进行调度，进行如下工作。每个 task 成员会记住他们之前的配置，并且在重新恢复时要求想提供的 slot。如果没有 slot 可用，task 将会向资源管理器请求一个新 slot。这种情况下，如果 TaskManager 没有任何可用的 slot，则该 task 就无法返回它之前的位置，而且不会将其他恢复的 task 赶出他们之前的 slot。之前的 slot 只能在 TaskManager 不再可用时消失，在这种情况下，一些 task 就需要请求一个新的 slot 了。调度策略可以赋予最大数量的 task 从本地状态恢复的机会，避免从其他 task 窃取他们之前的 slot。