Savepoint

Savepoint 和 Checkpoint 的区别

Savepoint 是命令触发的 Checkpoint，对流式程序做一次完整的快照并将结果写到 State backend，可用于停止、恢复或更新 Flink 程序。整个过程依赖于 Checkpoint 机制。另一个不同之处是，Savepoint 不会自动清除。

Checkpoint 的主要目的是为意外失败的作业提供恢复机制。 Checkpoint 的生命周期由 Flink 管理，无需用户交互。作为一种恢复和定期触发的方法，Checkpoint 实现有两个设计目标：i）轻量级创建和 ii）尽可能快地恢复。在用户终止作业后，通常会删除 Checkpoint（除非明确配置为保留的 Checkpoint）。

Savepoint 由用户创建，拥有和删除，用于手动备份和恢复或更新 Flink 程序。例如，升级 Flink 版本，调整用户逻辑，改变并行度等。 当然，Savepoint 需要在作业停止后继续存在。Savepoint 的生成，恢复成本可能更高一些，Savepoint 更多地关注可移植性和对前面提到的作业更改的支持。

除去概念上的差异，Checkpoint 和 Savepoint 的当前实现基本上使用相同的代码并生成相同的格式。然而有一个例外，可能会在未来引入更多的差异。例外情况是使用 RocksDB 状态后端的增量 Checkpoint。他们使用了一些 RocksDB 内部格式，而不是 Flink 的本机 Savepoint 格式。

分配 Operator IDs

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

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

DataStream<String> stream = env.  // Stateful source (e.g. Kafka) with ID  .addSource(new StatefulSource())  .uid("source-id") // ID for the source operator  .shuffle()  // Stateful mapper with ID  .map(new StatefulMapper())  .uid("mapper-id") // ID for the mapper  // Stateless printing sink  .print(); // Auto-generated ID

Savepoint 中会以 Operator ID 作为 key 保存每个有状态算子的状态

Operator ID | State------------+------------------------source-id   | State of StatefulSourcemapper-id   | State of StatefulMapper

Savepoint 操作

触发 Savepoint 时，会创建一个新的 Savepoint 目录，其中将存储数据和元数据。可以通过配置默认 targetDirectory 或指定自定义 targetDirectory。

# Default savepoint target directorystate.savepoints.dir: hdfs:///flink/savepoints

如果既未配置缺省值也未指定自定义目录，Savepoint 将失败。

从 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

$ bin/flink savepoint :jobId [:targetDirectory]

生成 Savepoint（以 jobId 作为唯一 ID），并返回创建的 Savepoint 的路径，需要此路径来还原和删除 Savepoint 。

在 Yarn 集群触发 Savepoint

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

要指定 jobId 和 yarnAppId（YARN 应用程序 ID），并返回创建的 Savepoint 的路径。

取消作业时生成 Savepoint

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

以原子方式触发具有 jobId 的 Savepoint，并取消作业。

恢复 Savepoint

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

提交作业，并指定要恢复的 Savepoint 路径。

允许启动有未恢复 State

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

默认情况下，恢复操作将尝试将 Savepoint 的所有 State 恢复。如果删除了算子，则可以通过 –allowNonRestoredState（简写为 -n） 选项跳过无法映射到新程序的状态。

删除 Savepoint

$ bin/flink savepoint -d :savepointPath

通过指定路径删除 Savepoint，也可以通过文件系统手动删除 Savepoint 数据，而不会影响其他 Savepoint 或 Checkpoint。

常见问题

应该为所有算子分配 ID 吗？

根据经验，是的。严格地说，只需要通过该 uid() 方法将 ID 分配给作业中的有状态算子。Savepoint 仅包含这些算子的 State，无状态算子不是保存点的一部分。

如果在作业中新添加一个有状态算子，会发生什么？

新算子将在没有任何状态的情况下进行初始化，类似于无状态算子。

如果在作业删除一个有状态的算子，会发生什么？

如果没有指定允许启动有未恢复 State（–allowNonRestoredState / -n），启动会失败。

如果在作业中重新排列有状态算子，会发生什么？

如果手动这些算子分配了 ID，作业将照常恢复。否则，重新排序后，有状态算子的自动生成 ID 很可能会更改，将导致无法从 Savepoint 恢复。

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

如果为有状态算子手动分配了 ID，作业将照常恢复，则无状态算子的改变不会影响。否则，重新排序后，有状态算子的自动生成 ID 很可能会更改，将导致无法从 Savepoint 恢复。

如果作业的并行性发生改变，会发生什么？

如果 Savepoint 的生成是使用 Flink 1.2.0 以及之后的版本，并且没有使用弃用状态 API，可以正常恢复作业。

如果 Savepoint 的生成比 Flink 1.2.0 更早的版本，或者使用弃用状态 API，则首先必须将作业和 Savepoint 升级到 1.2.0 以及之后的版本，然后才能更改并行度。请参考官方 升级指南。

Restart

当 Task 发生故障时，Flink 需要重启出错的 Task 以及其他受到影响的 Task ，以使得作业恢复到正常执行状态。

Flink 通过重启策略和故障恢复策略来控制 Task 重启：重启策略决定是否可以重启以及重启的间隔；故障恢复策略决定哪些 Task 需要重启。

Flink 支持多种不同的重启策略，控制着作业失败后如何重启。集群可以设置默认的重启策略，作业提交的时候也可以指定重启策略，覆盖默认的重启策略。

默认的重启策略配置在 conf/flink-conf.yaml，参数 restart-strategy 定义了采用什么策略。如果 checkpoint 未启用，就会采用 "no restart" 策略，如果启用了 checkpoint 机制，但是未指定重启策略的话，就会采用 "fixed-delay" 策略。每个重启策略都有自己的参数来控制它的行为，这些值也可以在配置文件中设置，每个重启策略的描述都包含着各自的配置值信息。

以下是支持的三种重启策略的可配置项：none、fixed-delay、failure-rate、exponential-delay（1.13.x）

除了定义一个默认的重启策略之外，你还可以为每一个 Job 指定它自己的重启策略，这个重启策略可以在 ExecutionEnvironment 中调用 setRestartStrategy() 方法来程序化地调用，这种方式同样适用于 StreamExecutionEnvironment。

固定延迟重启策略（Fixed Delay Restart Strategy）

尝试一个给定的次数来重启 Job，如果超过了最大的重启次数，Job 最终将失败。在连续的两次重启尝试之间，重启策略会等待一个固定的时间。

flink-conf.yaml 参数配置：

restart-strategy: fixed-delay
# Flink 尝试执行的次数，默认值：1restart-strategy.fixed-delay.attempts: 3
# 两次重启之间等待的时间，默认值：1srestart-strategy.fixed-delay.delay: 10 s

程序设置：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();env.setRestartStrategy(RestartStrategies.fixedDelayRestart(  3, // 尝试重启的次数  Time.of(10, TimeUnit.SECONDS) // 等待的时间));

失败率重启策略（Failure Rate Restart Strategy）

Job 失败后会重启次数如果超过失败率，Job 会最终被认定失败。在两个连续的重启尝试之间，重启策略会等待一个固定的时间。

flink-conf.yaml 参数配置：

restart-strategy:failure-rate
# Flink尝试执行的次数，默认值：1restart-strategy.failure-rate.max-failures-per-interval: 3 
# 计算失败率的时间间隔，默认值：1 minrestart-strategy.failure-rate.failure-rate-interval: 5 min 
# 两次重启之间等待的时间，默认值：1srestart-strategy.failure-rate.delay: 10 s

程序设置：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();env.setRestartStrategy(RestartStrategies.failureRateRestart(  3, // 每个时间间隔的最大故障次数  Time.of(5, TimeUnit.MINUTES), // 测量故障率的时间间隔  Time.of(10, TimeUnit.SECONDS) // 等待的时间));

指数延迟重启策略（Exponential Delay Restart Strategy）

试图无限地重新启动 Job，并将延迟指数增加到最大延迟，并延迟保持在最大值。

flink-conf.yaml 参数配置：

restart-strategy: exponential-delay
# 初始延迟，默认值：1srestart-strategy.exponential-delay.initial-backoff: 10 s
# 最大延迟，默认值：5minrestart-strategy.exponential-delay.max-backoff: 2 min
# 每次失败后，延迟时间乘以此值作为新的延迟时间，直到达到最大延迟，默认值：2.0restart-strategy.exponential-delay.backoff-multiplier: 2.0
# 作业运行多长时间后，延迟时间恢复为初始延迟，默认值：1hrestart-strategy.exponential-delay.reset-backoff-threshold: 10 min
# 随机偏移值，防止多个 job 同时重启，默认值：0.1restart-strategy.exponential-delay.jitter-factor: 0.1

程序设置：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();env.setRestartStrategy(RestartStrategies.exponentialDelayRestart(  Time.milliseconds(1),  Time.milliseconds(1000),  1.1,   Time.milliseconds(2000),   0.1 ));

无重启策略（No Restart Strategy）

Job 直接失败，不会尝试进行重启

flink-conf.yaml 参数配置：

restart-strategy: none

程序设置：

val env = ExecutionEnvironment.getExecutionEnvironment()env.setRestartStrategy(RestartStrategies.noRestart())

Failover Strategies

Flink 支持多种不同的故障恢复策略，该策略需要通过 Flink 配置文件 flink-conf.yaml 中的 jobmanager.execution.failover-strategy 配置项进行配置。

• full - 全部重启

• region - 局部重启

Restart All Failover Strategy

在“全部重启”恢复策略下，Task 发生故障时会重启作业中的所有 Task 进行故障恢复。

Restart Pipelined Region Failover Strategy

该策略会将作业中的所有 Task 划分为数个 Region。当有 Task 发生故障时，会尝试找出进行故障恢复需要重启的最小 Region 集合。

相比于“全部重启”恢复策略，这种策略在一些场景下的故障恢复需要重启的 Task 会更少。

此处 Region 指以 Pipelined 形式进行数据交换的 Task 集合。也就是说，Batch 形式的数据交换会构成 Region 的边界。

• DataStream 和 流式 Table/SQL 作业的所有数据交换都是 Pipelined 形式的。

• 批处理式 Table/SQL 作业的所有数据交换默认都是 Batch 形式的。

• DataSet 作业中的数据交换形式会根据 ExecutionConfig 中配置的 ExecutionMode 决定。

需要重启的 Region 的判断逻辑如下：

1. 出错 Task 所在 Region 需要重启。

2. 如果要重启的 Region 需要消费的数据有部分无法访问（丢失或损坏），产出该部分数据的 Region 也需要重启。

3. 需要重启的 Region 的下游 Region 也需要重启。这是出于保障数据一致性的考虑，因为一些非确定性的计算或者分发会导致同一个 Result Partition 每次产生时包含的数据都不相同。

[1]  https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/state/savepoints/ 

[2]  https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/execution/task_failure_recovery/