Flink状态管理与容错（二）

状态后端(StateBackend)

当Flink程序进行checkpoint检查点保存时，检查点的保存需要JobManager和TaskManager以及外部存储系统的协调。首先，JobManager会向所有的TaskManager发送触发检查点的命令，接收到命令的TaskManager会对当前任务的所有状态进行快照保存，并将其持久化到远程的存储介质中，完成保存后，TaskManager会向JobManager返回确认信息。整个过程是分布式的，只有当JobManager收到所有TaskManager的确认信息后，才会确认当前检查点成功保存。

checkpoint 检查点整个流程如下:

以上检查点保存的整个流程都由状态后端（StateBackend）协调和管理，Flink状态后端（StateBackend）是Flink中负责状态管理、存储和访问的可插拔组件，在检查点（checkpoint）的保存过程中起着重要的作用，状态后端主要有2个职责：

• TaskManager本地状态管理：状态后端负责管理应用程序在每个TaskManager本地的状态，它提供了状态的存储和访问接口，以便应用程序可以读取和更新状态。

• 检查点写入远程持久化存储：在检查点保存过程中，状态后端将任务的状态数据传输到指定的远程存储介质中，以确保状态的持久性和容错性。

状态后端分类

Flink提供了两类的状态后端，一类是HashMapStateBackend，另一类是EmbeddedRocksDBSatateBackend。

• HashMapStateBackend

HashMapStateBackend将状态数据以HashMap数据结构进行存储，默认将状态存储在JobManager内存中。通过用户指定checkpint持久化目录也可以将状态数据存储在外部持久化系统中。这种状态后端每次进行checkpoint检查点时都是全量方式进行，适用于较小的状态数据集，并且对于低延迟和高吞吐量的应用程序非常有效。

• EmbeddedRocksDBStateBackend

EmbeddedRocksDBStateBackend 是 Flink 的一种基于 RocksDB 的状态后端。RocksDB 是一个高性能、持久化的键值存储引擎，它将状态数据存储在本地磁盘上，默认在TaskManager本地数据目录中。与 HashMapStateBackend 状态存储数据结构不同，EmbeddedRocksDBStateBackend 数据以序列化的字节数组形式存储，读写操作需要序列化与发序列化，状态访问性能可能差一些，但RockDBStateBackend是目前唯一支持增量检查点的状态后端，可以保存非常大的状态，生产环境中建议使用这种方式存储状态。

针对以上两类状态后端，Flink提供了3中不同StateBackend状态后端实现，包括基于内存的MemoryStateBackend、基于文件系统的FsStateBackend，以及基于RockDB作为存储介质的RocksDBStateBackend，可以适应各种不同的应用场景和需求，下面分别介绍。

MemoryStateBackend

基于内存的状态管理具有非常快速和高效的特点，但也具有非常多的限制，最主要的就是内存的容量限制，一旦存储的状态数据过多就会导致系统内存溢出等问题，从而影响整个应用的正常运行。同时如果机器出现问题，整个主机内存中的状态数据都会丢失，进而无法恢复任务中的状态数据。因此从数据安全的角度建议用户尽可能地避免在生产环境中使用MemoryStateBackend。

代码使用方式如下：

//设置状态后端为HashMapStateBackend，状态数据存储在JobManager内存中
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

FsStateBackend

和MemoryStateBackend有所不同，FsStateBackend是基于文件系统的一种状态管理器，这里的文件系统可以是本地文件系统，也可以是HDFS分布式文件系统。FsStateBackend更适合任务状态非常大的情况，例如应用中含有时间范围非常长的窗口计算，或Key/value State状态数据量非常大的场景。

代码使用方式如下：

//设置状态后端为HashMapStateBackend，状态数据存储在本地文件系统中
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");

RocksDBStateBackend

RocksDBStateBackend是Flink中内置的第三方状态管理器，和前面的状态管理器不同，RocksDBStateBackend需要单独引入相关的依赖包到工程中，Java代码和Scala代码中都需要导入。

<!– Flink Rocksdb 状态后端 依赖包 –>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>${flink.version}</version>
</dependency>

RocksDBStateBackend采用增量方式进行状态数据的Snapshot，与FsStateBackend相比，虽说都是将状态存储在磁盘中，RocksDBStateBackend在性能上要比FsStateBackend高一些，和MemoryStateBackend相比性能就会较弱一些。RocksDB克服了State受内存限制的缺点，同时又能够将状态增量持久化到远端文件系统中，推荐在生产中使用。

代码使用方式如下：

//设置状态后端为RocksDBStateBackend，状态数据存储在本地文件系统中
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");

状态后端全局配置

以上两类状态后端的三种实现方式除了可以在代码中进行配置外，还可以在Flink集群JobManager FLINK_HOME/conf/flink-conf.yaml配置文件中配置,从而做到状态后端全局配置。配置分别如下：

• MemoryStateBackend

state.backend: hashmap
state.checkpoint-storage: jobmanager

• FsStateBackend

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

• RocksDBStateBackend

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

默认情况下，JobManager节点的flink-conf.yaml配置文件如果设置了checkpoint选项，则Flink只保留最近成功生成的1个checkpoint，而当Flink程序失败时，可以通过最近的checkpoint来进行恢复。但是，如果希望保留多个checkpoint，并能够根据实际需要选择其中一个进行恢复，就会更加灵活。添加如下配置，指定最多可以保存的checkpoint的个数。

state.checkpoints.num-retained: 2

状态后端代码案例

下面以将Flink状态数据存储在HDFS中，使用状态后端管理为rocksdb编写Flink代码，来演示Flink状态后端全局配置及使用，这里以Flink 向Yarn中Application模式提交任务为例进行演示。

1) 配置flink-conf.yaml

在node5节点的FLINK_HOME/conf/flink-conf.yaml配置文件“ Fault tolerance and checkpointing”模块下配置如下内容。

state.backend.type: rocksdb
state.checkpoints.dir: hdfs://mycluster/flink-checkpoints
state.checkpoint-storage: filesystem
state.checkpoints.num-retained: 2

2) 编写Flink 读取socket代码并打包

• Java代码：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//开启checkpoint，每隔1000ms进行一次checkpoint
env.enableCheckpointing(1000);
​
//设置checkpoint清理策略为RETAIN_ON_CANCELLATION
env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
​
/**
* Socket输入数据如下：
* hello,flink
* hello,flink
* hello,rocksdb
*/
DataStreamSource<String> ds = env.socketTextStream("node5", 9999);
ds.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
  @Override
  public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
      String[] words = s.split(",");
      for (String word : words) {
          collector.collect(new Tuple2<>(word,1));
      }
  }
}).keyBy(one->one.f0).sum(1).print();
​
env.execute();

• scala代码：

val env = StreamExecutionEnvironment.getExecutionEnvironment
//导入隐式转换
import org.apache.flink.streaming.api.scala._
​
// 开启checkpoint，每隔1000ms进行一次checkpoint
env.enableCheckpointing(1000)
​
// 设置checkpoint清理策略为RETAIN_ON_CANCELLATION
env.getCheckpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
​
/**
* Socket输入数据如下：
* hello,flink
* hello,flink
* hello,rocksdb
*/
val ds: DataStream[String] = env.socketTextStream("node5", 9999)
​
ds.flatMap(_.split(","))
.map((_, 1))
.keyBy(_._1)
.sum(1)
.print()
​
env.execute()

以上无论是Java代码还是Scala代码编写完成后，进行打包，上传到node5节点的/root/flink-jar-test目录中。

3) 启动HDFS

#启动Zookeeper集群
[root@node3 ~]# zkServer.sh start
[root@node4 ~]# zkServer.sh start
[root@node5 ~]# zkServer.sh start
​
#启动HDFS集群和Yarn集群
[root@node1 ~]# start-all.sh

注意：启动HDFS集群后，如果有hdfs://mycluster/flink-checkpoints目录，为了后续方便看出任务状态目录，最好删除该目录。

4) 提交任务并输入数据

在node5节点上向Yarn中提交Flink任务，这里以提交Scala任务为例。

#提交任务之前在node5节点启动Socket服务
[root@node5 ~]# nc -lk 9999
​
#提交Flink任务
[root@node5 ~]# cd /software/flink-1.17.1/bin/
[root@node5 bin]# ./flink run-application -t yarn-application -c com.mashibing.flinkscala.code.chapter7.checkpoints.RocksDBStateBackendTest /root/flink-jar-test/FlinkScalaCode-1.0-SNAPSHOT-jar-with-dependencies.jar

以上任务提交执行后，向Socket中输入如下数据:

hello,flink
hello,flink
hello,rocksdb

观察Flink任务状态统计的结果:

5) 取消任务并状态恢复

在Flink WebUI中取消Flink任务，然后查看HDFS中存储checkpoint对应的状态目录：

重新在node5节点提交Flink任务，通过参数-s 指定任务对应的状态目录恢复状态：

[root@node5 bin]# ./flink run-application -t yarn-application -s hdfs://mycluster/flink-checkpoints/5bc3f216025c57bc311338482bceeea2/chk-86 -c com.mashibing.flinkscala.code.chapter7.checkpoints.RocksDBStateBackendTest /root/flink-jar-test/FlinkScalaCode-1.0-SNAPSHOT-jar-with-dependencies.jar

提交任务后，继续向socket服务器中输入如下数据：

hello,flink
hello,rocksdb

观察Flink 新启动任务的WebUI中TaskManager中统计的状态如下，可以看到状态成功被恢复回来。

保存点(savepoint)

savepoints 是检查点的一种特殊实现，底层实现其实也是使用checkpoints的机制。savepoints是用户以手工命令的方式触发checkpoint,并将结果持久化到指定的存储路径中，其主要目的是帮助用户在升级和维护集群过程中保存系统中的状态数据，避免因为停机运维或者升级应用等正常终止应用的操作而导致系统无法恢复到原有的计算状态的情况，从而无法实现从端到端的 exactly-once语义保证。

savepoint配置

• 配置savepoint 的存储路径

我们可以在flink-conf.yaml中配置savepoint存储的位置，设置后，如果要创建指定Job的savepoint，可以不用在手动执行命令时指定savepoint的位置。

state.savepoints.dir: hdfs://mycluster/savepoints

• 在代码中设置算子ID

为了能够在作业的不同版本之间以及Flink的不同版本之间顺利升级，强烈推荐程序员通过手动给算子赋予ID，这些ID将用于确定每一个算子的状态范围。如果不手动给各算子指定ID，则会由Flink自动给每个算子生成一个ID。而这些自动生成的ID依赖于程序的结构，并且对代码的更改是很敏感的。因此，强烈建议用户手动设置ID。

env.socketTextStream("node5",9999).uid("socket-source")
      .flatMap((String line,Collector<Tuple2<String,Integer>> out) -> {
          String[] words = line.split(",");
          for (String word : words) {
              out.collect(new Tuple2<>(word, 1));
          }
      }).returns(Types.TUPLE(Types.STRING,Types.INT)).uid("flatmap")
      .keyBy(tp -> tp.f0)
      .sum(1).uid("sum")
      .print().uid("print");

savepoint 使用案例

这里以读取Socket中的数据进行WordCount为例来演示savepoint的配置及使用，可以按照如下步骤进行测试。

1) 编写代码并打包

• java代码

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
​
//读取socket数据做wordcount
SingleOutputStreamOperator<String> lines = env.socketTextStream("node5", 9999).uid("socket-source");
​
SingleOutputStreamOperator<Tuple2<String, Integer>> tuple2 = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
  @Override
  public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
      String[] words = s.split(",");
      for (String word : words) {
          collector.collect(new Tuple2<>(word, 1));
      }
  }
}).uid("flatmap");
​
SingleOutputStreamOperator<Tuple2<String, Integer>> result = tuple2.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
  @Override
  public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
      return stringIntegerTuple2.f0;
  }
}).sum(1).uid("sum");
​
result.print().uid("print");
​
env.execute();

• scala代码

val env = StreamExecutionEnvironment.getExecutionEnvironment

//导入隐式转换
import org.apache.flink.streaming.api.scala._

// 读取socket数据做wordcount
env.socketTextStream("node5", 9999).uid("socket-source")
.flatMap(_.split(" ")).uid("flatMap")
.map((_, 1)).uid("map")
.keyBy(_._1)
.sum(1).uid("sum")
.print().uid("print")

env.execute()

以上代码编写完成后，打包并上传到node5节点的/root/flink-jar-test目录中。

2) 在flink-conf.yaml中配置savepoint路径

在node5节点配置flink-conf.yaml文件“Fault tolerance and checkpointing”部分配置savepoint保存的路径，如下。

state.savepoints.dir: hdfs://mycluster/flink-savepoints

3) 启动HDFS集群

#启动Zookeeper集群
[root@node3 ~]# zkServer.sh start
[root@node4 ~]# zkServer.sh start
[root@node5 ~]# zkServer.sh start
​
#启动HDFS集群和Yarn集群
[root@node1 ~]# start-all.sh

注意：启动HDFS集群后，如果有hdfs://mycluster/flink-savepoints目录，为了后续方便看出任务状态目录，最好删除该目录。

4) 提交任务并统计状态结果

node5节点启动socket服务并在node5节点上提交Flink任务，提交Flink任务后，向Socket中输入数据，观察Flink任务webui中统计的结果。

#node5节点启动socket服务
[root@node5 ~]# nc -lk 9999
​
#node5节点提交Flink任务
[root@node5 conf]# cd /software/flink-1.17.1/bin/
[root@node5 bin]# ./flink run-application -t yarn-application -c com.mashibing.flinkjava.code.chapter7.savepoints.SavePointTest /root/flink-jar-test/FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar

提交任务后，向socket中输入以下数据：

hello,flink
hello,flink
hello,savepoint

输入数据后，观察Flink WebUI统计的结果如下:

5) 触发savepoint

执行如下命令执行savepoint 操作，将当前Flink程序的状态保存到对应路径中。

[root@node5 bin]# ./flink savepoint d2d863913b663ad70f60ee50d05aaa32 -yid application_1687674094408_0001

以上命令注意以下几点：

• savepoint操作命令格式为：./flink savepoint <Job ID> [<target directory>]

• 如果savepoint path(target directory)在当前提交任务节点的flink-conf.yaml中配置了，就不需要再写上。

• 如果是基于Yarn中运行的Flink任务，Flink JobID 通过Flink WebUI查看，并且执行savepoint命令最后需要通过-yid参数指定yarn application ID 链接到Yarn Application中。

以上savepoint 命令执行后，可以看到HDFS中生成对应的savepoint路径：

此时，我们可以手动取消Flink任务或者通过命令停止Flink任务，操作如下:

#命令方式取消Flink任务
[root@node5 bin]# ./flink cancel d2d863913b663ad70f60ee50d05aaa32 -yid application_1687674094408_0001

6) 从savepoint启动Job

可以通过如下命令来从savepoint路径中恢复Flink Job状态，命令如下：

[root@node5 bin]# ./flink run-application -t yarn-application -s hdfs://mycluster/flink-savepoints/savepoint-d2d863-0dd2aae39a79 -c com.mashibing.flinkjava.code.chapter7.savepoints.SavePointTest /root/flink-jar-test/FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar

Flink任务启动后，可以继续向Socket中输入如下数据：

hello,flink
hello,savepoint

可以观察Flink WebUI对应统计的状态结果，可以看到Flink 任务已经成功从savepoint中恢复过来。

Checkpoint & Savepoint区别

Checkpoint和Savepoint是Apache Flink中用于状态管理和故障恢复的两个概念。

Checkpoint是周期性自动创建的，用于将应用程序的状态信息保存到持久化存储中。它记录了输入数据、中间计算结果和应用程序配置等状态，以便在故障发生时能够从最近的Checkpoint恢复应用程序的状态。

Savepoint是由用户手动触发的操作，用于显式地保存应用程序的状态。用户可以在任意时间点创建Savepoint，并将应用程序的状态保存到持久化存储中。Savepoint可以用于应用程序升级、在不同环境中迁移应用程序的状态。

综上所述，Flink中的Checkpoint是周期性自动创建的，用于实现故障恢复。它将应用程序的状态信息写入持久化存储，以便在故障发生时可以恢复到先前的状态。而Savepoint是由用户手动触发的操作，用于保存应用程序状态以供以后使用,Savepoint可以用于状态迁移、版本回滚/升级等场景。

Flink端到端一致性保证

本小节主要介绍数据处理的一致性语义，以及Flink中实现端到端数据一致性保证的方式和原理。

数据处理一致性语义

分布式系统中常用这么几种语义（semantics）来描述系统在经历了故障恢复后，内部各个组件之间状态的一致性。严格程度从高到低为：exactly once（准确一次）、at least once（至少一次）、at most once（最多一次）。下面以一个例子分别解释以上三种一致性语义的界定。

数据源S发送给算子A数据1,2,3,4,5…，由A进行累加，也就是sum。A中的状态就是累加的和，S的状态就是当前已发送的流（因为A的当前结果和S之前发送的所有数据都有关）。一开始，S发出了1,2,3。A的状态依次更新为1,3,6。当S发出4的时候，A出现了故障。现在S和A的状态分别是S：1,2,3,4，A:6。

接下来，当整个程序从故障中恢复过程中可能有这么几种情况:

• at most once (最多消费一次)

S不知道A发生了故障，随后系统从故障中恢复，A收到了S发送的下一条数据“5”，数据“4”实际上被丢失了。这时S的发送历史是1,2,3,4,5，A的状态是11。而“1,2,3,4,5”所对应的累加和应该是15，所以S和A就出现了状态不一致，S认为A的状态包含了“4”，而A实际上没有。这样就导致“4”这条数据永久丢失，这种程序从故障恢复的语义就是at most once 语义。

• exactly once (精准消费一次)

回到状态S：1,2,3,4，A：6的时间点，假设这时S知道A有故障，并且A没有处理“4”，系统从故障中恢复之后，S和A都退回到上一个一致的状态，然后重新开始运算。比如上一个一致的状态是：S:1,2,3, A:6，那么S回退之后会重发“4”。如果A成功处理“4”，现在的状态就变成：S：1,2,3,4， A：10。S和A的状态就是一致的。如果上一个一致的状态是：S:1,2, A:3，那么S回退之后会从“3”开始重发。这种程序从故障恢复的语义就是exactly once 语义。

以上这个过程我们假定的是A没有处理“4”这条数据，那么还有可能是A已经处理了“4”，然后A在通知S“4已处理”之前挂了，这时S不知道A已经处理了“4”，所以为了确保一致性，在恢复程序时，S必须假设A没有处理"4"，因此S和A仍然要一起退回上一个一致的状态，并且从一致的状态重新开始运算。比如说：S:1,2,3, A:6。S重发“4”并且被A处理之后，虽然A实际上处理了“4”两次，但A第一次处理“4”之后的状态在故障之后弃用了。所以在A当前的状态中，“4”只被处理了一次，这也是exactly once处理语义。

• at least once(至少消费一次)

同样回到状态S：1,2,3,4，A：6的时间点，但是S和A没有一起退回上一个一致的状态，比如说在程序恢复之后，两者的状态为：S:1,2,3, A: 10。这时，S重发“4”，状态变为S:1,2,3,4 A: 14，当前状态下，“4”被处理了两次，且状态不一致。这种程序从故障恢复的语义就是at least once 语义。

通过以上的了解，我们发现exactly-once语义中无论一条数据被重复处理多少次，只要影响最后结果是一次，那么这种我们也称为是exactly-once语义保证，这个就是所谓的幂等(idempotent)操作（执行多次与执行一次的效果相同）。

Flink端到端数据处理一致性

在Flink内部，Flink通过checkpoint检查点机制来保证Flink内部处理数据的一致性，一般的流处理系统提供了程序内部的数据处理一致性，但不包括输入和输出的一致性。在Flink中，数据处理一致性指的是端到端的一致性，包括输入、系统内部和输出的整体一致性。因此，为了实现端到端的一致性保证，除了保证Flink内部数据处理一致性外，还需要保证Flink读取数据的输入端和数据输出端的一致性。

输入端的一致性取决于外部数据源能否按照不变的顺序重放数据。针对Flink输入端想要保证Flink程序故障恢复不丢失数据，数据输入端必须具备数据重放的能力。例如，Flink读取Socket数据时，socket数据流不具备数据重放功能，Flink程序从故障恢复时就不能保证exactly-once 处理数据的语义，读取Socket中的数据的保证语义只能是at most once。再如，Flink读取Kafka中数据或者文件数据时，Flink可以将读取source的偏移量作为状态存储在checkpoint中，就可以保证Flink程序故障恢复时exactly-once语义保证，所以只要数据源可以保证数据重放，读取数据源的offset又能被Flink以状态进行保存，保证输入端的数据一致性就相对简单。

输出端数据一致性的保证相对复杂。在Flink内部，通过状态回滚可以保证Flink程序故障重启后的exactly-once数据处理，即使在进行checkpoint之前某些数据已被处理，但只要它们还没有进行checkpoint，故障重启后的Flink程序会将这部分数据视为未被处理（因为状态已回滚），从而不会对结果产生影响。然而，如果涉及将数据写出到外部系统，那么在Flink程序挂掉之前，这部分数据可能已经被写出到外部系统，当Flink程序故障重启后，这部分数据将被重新处理一次，导致数据重复处理，即出现at-least-once处理语义。

因此，为了保证Flink程序故障恢复后输出端的exactly-once处理语义，输出端必须具备幂等性（idempotency）或支持事务（transaction）。

幂等性指的是重复数据写入外部系统，但只对结果产生一次影响。例如，根据主键向MySQL或根据Rowkey向HBase写入数据时，重复数据写入只会对结果产生一次影响。然而，幂等性方式需要外部写入目标支持幂等写入，因此受限场景较多。另一种方式是使用事务来保证输出端数据的一致性。

使用事务方式保证输出端数据一致性的主要思路是，在Flink程序向外部系统Sink数据时，构建一个代码事务：当遇到Flink barrier时，创建一个新的事务，并将后续所有写出的数据绑定到该事务。当Sink收到Flink JobManager 发来的Checkpoint检查点完成通知时，提交该事务，该事务对应的数据才真正写入到外部系统中。如果只是创建了新的事务而Flink出现故障，当Flink程序故障恢复后，该事务由于没有提交，所以对应的数据不会写入到外部系统。通过这种事务的方式，数据写出操作实现了随着checkpoint进行提交和回滚。实际上，这种事务方式可以理解为Flink将外部系统写出部分视为其内部状态管理的一部分。

为了实现这种事务方式，Flink提供了TwoPhaseCommitSinkFunction抽象类，方便自定义实现两阶段提交方式来写出数据，以保证输出端数据的一致性。关于两阶段提交的具体细节，可以参考Flink写出Kafka的exactly-once保证相关小节。

Flink写出Kafka exactly once 保证

本小节对Flink写出Kafka exactly once 语义保证实现原理进行介绍，同时会代码实现Flink TwoPhaseCommitSinkFunction 两阶段提交，学习如何手动干预端到端语义一致性保证。

Flink消费Kafka 数据offset提交配置

Flink提供了消费kafka数据的offset如何提交给Kafka或者zookeeper(kafka0.8之前)的配置。注意，Flink并不依赖提交给Kafka或者zookeeper中的offset来保证容错。提交的offset只是为了外部来查询监视kafka数据消费的情况。

配置offset的提交方式取决于是否为job设置开启checkpoint。可以使用env.enableCheckpointing(5000)来设置开启checkpoint。

• 关闭checkpoint：

如果禁用了checkpoint，那么offset位置的提交取决于Flink读取kafka客户端的配置，enable.auto.commit ( auto.commit.enable【Kafka 0.8】，默认false)配置是否开启自动提交offset, auto.commit.interval.ms(默认5s)决定自动提交offset的周期。

• 开启checkpoint：

如果开启了checkpoint，那么当checkpoint保存状态完成后，将checkpoint中保存的offset位置提交到kafka。这样保证了Kafka中保存的offset和checkpoint中保存的offset一致，Flink1.14版本后，enable.auto.commit改参数默认为false，所以开启了checkpoint后，checkpoint中保存的offset数据不会自动向Kafka提交，如果需要提交可以设置改参数为true。

checkpoint+两阶段提交保证exactly once语义

当谈及“exactly-once semantics”仅一次处理数据时，指的是每条数据只会影响最终结果一次。Flink可以保证当机器出现故障或者程序出现错误时，也没有重复的数据或者未被处理的数据出现，实现仅一次处理的语义。Flink开发出了checkpointing机制，这种机制是在Flink应用内部实现仅一次处理数据的基础。

checkpoint中包含：

• 当前应用的状态

• 当前消费流数据的位置

在Flink1.4版本之前，Flink仅一次处理数据只限于Flink应用内部（可以使用checkpoint机制实现仅一次数据数据语义），当Flink处理完的数据需要写入外部系统时，不保证仅一次处理数据。为了提供端到端的仅一次处理数据，在将数据写入外部系统时也要保证仅一次处理数据，这些外部系统必须提供一种手段来允许程序提交或者回滚写入操作，同时还要保证与Flink的checkpoint机制协调使用。

在分布式系统中协调提交和回滚的常见方法就是两阶段提交协议。下面给出从kafka中读取数据，经过处理数据之后将结果再写回kafka的实例了解Flink如何使用两阶段提交协议来实现数据仅一次处理语义。

kafka0.11版本之后支持事务，这也是Flink与kafka交互时仅一次处理的必要条件。【注意：当Flink处理完的数据写入kafka时，即当sink为kafka时，自动封装了两阶段提交协议】。Flink支持仅一次处理数据不仅仅限于和Kafka的结合，只要sink提供了必要的两阶段协调实现，可以对任何sink都能实现仅一次处理数据语义。

其原理如下：

上图Flink程序包含以下组件：

要使sink支持仅一次处理数据语义，必须以事务的方式将数据写往kafka,将两次checkpoint之间的操作当做一个事务提交，确保出现故障时操作能够被回滚。假设出现故障，在分布式多并发执行sink的应用程序中，仅仅执行单次提交或回滚事务是不够的，因为分布式中的各个sink程序都必须对这些提交或者回滚达成共识，这样才能保证两次checkpoint之间的数据得到一个一致性的结果。Flink使用两阶段提交协议(pre-commit+commit)来实现这个问题。

Filnk checkpointing开始时就进入到pre-commit阶段，具体来说，一旦checkpoint开始，Flink的JobManager向输入流中写入一个checkpoint barrier将流中所有消息分隔成属于本次checkpoint的消息以及属于下次checkpoint的消息，barrier也会在操作算子间流转，对于每个operator来说，该barrier会触发operator的State Backend来为当前的operator来打快照。如下图示：

Flink DataSource中存储着Kafka消费的offset，当完成快照保存后，将chechkpoint barrier传递给下一个operator。这种方式只有在Flink内部状态的场景是可行的，内部状态指的是由Flink的State Backend管理状态，例如上面的window的状态就是内部状态管理。只有当内部状态时，pre-commit阶段无需执行额外的操作，仅仅是写入一些定义好的状态变量即可，checkpoint成功时Flink负责提交这些状态写入，否则就不写入当前状态。

但是，一旦operator操作包含外部状态，事情就不一样了。我们不能像处理内部状态一样处理外部状态，因为外部状态涉及到与外部系统的交互。这种情况下，外部系统必须要支持可以与两阶段提交协议绑定的事务才能保证仅一次处理数据。

本例中的data sink是将数据写往kafka，因为写往kafka是有外部状态的，这种情况下，pre-commit阶段下data sink 在保存状态到State Backend的同时，还必须pre-commit外部的事务。如下图：

当checkpoint barrier在所有的operator都传递一遍切对应的快照都成功完成之后，pre-commit阶段才算完成。这个过程中所有创建的快照都被视为checkpoint的一部分，checkpoint中保存着整个应用的全局状态，当然也包含pre-commit阶段提交的外部状态。当程序出现崩溃时，我们可以回滚状态到最新已经完成快照的时间点。

下一步就是通知所有的operator，告诉它们checkpoint已经完成，这便是两阶段提交的第二个阶段：commit阶段。这个阶段中JobManager会为应用中的每个operator发起checkpoint已经完成的回调逻辑。本例中，DataSource和Winow操作都没有外部状态，因此在该阶段，这两个operator无需执行任何逻辑，但是Data Sink是有外部状态的，因此此时我们需要提交外部事务。如下图示：

汇总以上信息，总结得出：

• 一旦所有的operator完成各自的pre-commit,他们会发起一个commit操作。

• 如果一个operator的pre-commit失败，所有其他的operator 的pre-commit必须被终止，并且Flink会回滚到最近成功完成的checkpoint位置。

• 一旦pre-commit完成，必须要确保commit也要成功，内部的operator和外部的系统都要对此进行保证。假设commit失败【网络故障原因】，Flink程序就会崩溃，然后根据用户重启策略执行重启逻辑，重启之后会再次commit。

因此，所有的operator必须对checkpoint最终结果达成共识，即所有的operator都必须认定数据提交要么成功执行，要么被终止然后回滚。

两阶段提交代码案例

在Flink中TwoPhaseCommitSinkFunction是个抽象类，用户自定义类继承该类时需要实现如下5个方法:

• beginTransaction:开启一个事务，后续来的数据都属于该事务内的数据。

• invoke:当Sink中接收到一条数据时都会调用invoke方法，该方法中主要决定如何处理该数据。

• preCommit: pre-commit预提交阶段，当barrir到达后，会调用该方法进行数据预提交，同时调用beginTransaction开启一个新的事务执行属于下一个checkpoint的数据操作。

• commit：Flink checkpoint完成，真正执行commit提交操作，Flink在notifyCheckpointComplete()方法中调用该方法，即JobManager完成checkpoint后调用该方法。

• abort:一旦代码出现异常调用abort方法，该方法内进行终止事务，数据回滚操作。

下面以读取Kafka中数据写入到MySQL为例来演示用户自定义实现TwoPhaseCommitSinkFunction完成端到端数据处理一致性。

案例：读取Kafka中数据写入到MySQL中。

1) MySQL中创建表

use mydb;
create table mydb.user(
id int,
name varchar(255),
age int
);

2) Kafka中创建对应的topic

#启动Kafka后创建对应的topic
[root@node1 ~]# kafka-topics.sh –bootstrap-server node1:9092,node2:9092,node3:9092 –create –topic 2pc-topic –partitions 3 –replication-factor 3
​
[root@node1 ~]# kafka-topics.sh –bootstrap-server node1:9092,node2:9092,node3:9092 –list

3) 编写代码

• Java代码-JdbcCommonUtils

/**
* 连接MySQL – JDBC工具类
*/
public class JdbcCommonUtils{
  private static String driver = "com.mysql.jdbc.Driver";
  private static String url = "jdbc:mysql://node2:3306/mydb?useSSL=false";
  private static String username = "root";
  private static String password = "123456";
​
  private static Connection conn ;
​
  /**
    * 获取数据库连接对象
    */
  public Connection getConnect() {
      try {
          if(conn ==null){
              Class.forName(driver);
              conn = DriverManager.getConnection(url, username, password);
              //设置手动提交事务
              conn.setAutoCommit(false);
          }
      } catch (SQLException e) {
          throw new RuntimeException(e);
      } catch (ClassNotFoundException e) {
          throw new RuntimeException(e);
      }
      return conn;
  }
​
  /**
    * 提交事务
    */
  public void commit() {
      if (conn != null) {
          try {
              conn.commit();
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
​
      }
  }
​
  /**
    * 回滚事务
    */
  public void rollback() {
      if (conn != null) {
          try {
              conn.rollback();
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
​
      }
  }
​
  /**
    * 关闭连接
    */
  public void close() {
      if (conn != null) {
          try {
              conn.close();
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
​
      }
  }
}
 

• Java代码 – TwoPhaseCommitTest

/**
* 两阶段提交测试 – TwoPhaseCommitSinkFunction 类实现
* 案例：通过两阶段提交实现类完成读取Kafka数据写入到MySQL
*/
public class TwoPhaseCommitTest {
  public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.setParallelism(1);
​
      //开启checkpoint
      env.enableCheckpointing(5000);
​
      /**
        * Kafka中输入数据如下:
        * 1,zs,18
        * 2,ls,20
        * 3,ww,19
        * 4,zl,21
        * 5,tq,22
        */
      KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
              .setBootstrapServers("node1:9092,node2:9092,node3:9092") //设置Kafka 集群节点
              .setTopics("2pc-topic") //设置读取的topic
              .setGroupId("my-test-group") //设置消费者组
              .setStartingOffsets(OffsetsInitializer.latest()) //设置读取数据位置
              .setValueOnlyDeserializer(new SimpleStringSchema()) //设置value的序列化格式
              .build();
​
      DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(),
              "kafka-source");
​
      //自定义Sink 到MySQL,通过实现TwoPhaseCommitSinkFunction接口
      kafkaDS.addSink(new CustomTwoPhaseCommitSinkFunction());
​
      env.execute();
​
  }
​
}
​
/**
* 自定义 Sink 到 MySQL,通过继承TwoPhaseCommitSinkFunction抽象类
* TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
*     IN: 输入数据类型
*     TXN: 当前事务流程中的对象，贯穿TwoPhaseCommitSinkFunction类中各个方法中
*     CONTEXT: 上下文类型
*/
class CustomTwoPhaseCommitSinkFunction extends TwoPhaseCommitSinkFunction<String, JdbcCommonUtils, Void> {
​
  //创建默认的构造方法
  public CustomTwoPhaseCommitSinkFunction() {
      super(new KryoSerializer<>(JdbcCommonUtils.class,new ExecutionConfig()), VoidSerializer.INSTANCE);
  }
​
  /**
    * 开始事务。这里Flink会调用beginTransaction()方法创建连接池
    */
  @Override
  protected JdbcCommonUtils beginTransaction() throws Exception {
      System.out.println("beginTransaction()…");
      return new JdbcCommonUtils();
  }
​
  /**
    * 每接收一条数据后，会调用invoke()方法，将数据写入到MySQL中
    */
  @Override
  protected void invoke(JdbcCommonUtils jdbcUtils, String value, Context context) throws Exception {
      //将数据写入到MySQL中
      String[] split = value.split(",");
      PreparedStatement pst = jdbcUtils.getConnect().prepareStatement("insert into user(id,name,age) values(?,?,?)");
      pst.setInt(1,Integer.valueOf(split[0]));
      pst.setString(2,split[1]);
      pst.setInt(3,Integer.valueOf(split[2]));
​
      //执行插入操作
      pst.execute();
​
      //关闭pst对象
      pst.close();
  }
​
  /**
    * 当barrier 到达后，Flink会调用preCommit()方法，进行数据预提交
    * 预提交，如果一个preCommit执行失败，其他preCommit也会失败，Flink会调用abort()方法
    */
  @Override
  protected void preCommit(JdbcCommonUtils jdbcCommonUtils) throws Exception {
      System.out.println("barrier 到达，preCommit() 方法执行，开始预提交…");
      //这里的逻辑放在invoke()方法中进行插入数据
  }
​
  /**
    * Flink checkpoint完成，真正执行提交，Flink在notifyCheckpointComplete()方法中调用该方法，即JobManager完成checkpoint后调用该方法
    */
  @Override
  protected void commit(JdbcCommonUtils jdbcCommonUtils) {
      System.out.println("commit() 方法执行…");
      //提交事务
      jdbcCommonUtils.commit();
  }
​
  /**
    * 代码出现异常，事务中止，Flink会调用abort()方法
    * 这里主要是回滚事务
    */
  @Override
  protected void abort(JdbcCommonUtils jdbcCommonUtils) {
      System.out.println("abort() 方法执行…");
      //回滚事务
      jdbcCommonUtils.rollback();
      //关闭连接
      jdbcCommonUtils.close();
  }
}

• Scala代码 – JdbcCommonUtils

/**
* 连接Mysql – JDBC工具类
* Scala中，类被替换为了对象，并且不再需要使用static关键字
*/
case class JdbcCommonUtils() {
val driver = "com.mysql.jdbc.Driver"
val url = "jdbc:mysql://node2:3306/mydb?useSSL=false"
val username = "root"
val password = "123456"
​
private var conn: Connection = _
/**
  * 获取数据库连接对象
  */
def getConnect: Connection = {
  if (conn == null) {
    try {
      Class.forName(driver)
      conn = DriverManager.getConnection(url, username, password)
      //设置手动提交事务
      conn.setAutoCommit(false)
    } catch {
      case e: SQLException =>
        throw new RuntimeException(e)
      case e: ClassNotFoundException =>
        throw new RuntimeException(e)
    }
  }
  conn
}
​
/**
  * 提交事务
  */
def commit(): Unit = {
  if (conn != null) {
    try {
      conn.commit()
    } catch {
      case e: SQLException =>
        throw new RuntimeException(e)
    }
  }
}
​
/**
  * 回滚事务
  */
def rollback(): Unit = {
  if (conn != null) {
    try {
      conn.rollback()
    } catch {
      case e: SQLException =>
        throw new RuntimeException(e)
    }
  }
}
​
/**
  * 关闭连接
  */
def close(): Unit = {
  if (conn != null) {
    try {
      conn.close()
    } catch {
      case e: SQLException =>
        throw new RuntimeException(e)
    }
  }
}
}

• Scala代码-TwoPhaseCommitTest

/**
* 两阶段提交测试 – TwoPhaseCommitSinkFunction 类实现
* 案例：通过两阶段提交实现类完成读取Kafka数据写入到MySQL
*/
object TwoPhaseCommitTest {
def main(args: Array[String]): Unit = {
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  //开启隐式转换
  import org.apache.flink.streaming.api.scala._
​
  //设置并行度
  env.setParallelism(3)
​
  //开启checkpoint
  env.enableCheckpointing(5000)
​
  /**
    * Kafka中输入数据如下:
    * 1,zs,18
    * 2,ls,20
    * 3,ww,19
    * 4,zl,21
    * 5,tq,22
    */
  val kafkaSource: KafkaSource[String] = KafkaSource.builder()
    .setBootstrapServers("node1:9092,node2:9092,node3:9092") //设置Kafka 集群节点
    .setTopics("2pc-topic") //设置读取的topic
    .setGroupId("my-test-group") //设置消费者组
    .setStartingOffsets(OffsetsInitializer.latest()) //设置读取数据位置
    .setValueOnlyDeserializer(new SimpleStringSchema()) //设置value的序列化格式
    .build()
​
  val kafkaDS: DataStream[String] = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(),
    "kafka-source")
​
  //自定义Sink 到MySQL,通过实现TwoPhaseCommitSinkFunction接口
  kafkaDS.addSink(new CustomTwoPhaseCommitSinkFunction)
​
  env.execute()
}
​
}
​
/**
* 自定义 Sink 到 MySQL,通过继承TwoPhaseCommitSinkFunction抽象类
* TwoPhaseCommitSinkFunction[IN, TXN, CONTEXT]
*     IN: 输入数据类型
*     TXN: 当前事务流程中的对象，贯穿TwoPhaseCommitSinkFunction类中各个方法中
*     CONTEXT: 上下文类型
*/
class CustomTwoPhaseCommitSinkFunction extends TwoPhaseCommitSinkFunction[String, JdbcCommonUtils, Void](
new KryoSerializer[JdbcCommonUtils](classOf[JdbcCommonUtils], new ExecutionConfig),
VoidSerializer.INSTANCE
) {
/**
  * 开始事务。这里Flink会调用beginTransaction()方法创建连接池
  */
override def beginTransaction(): JdbcCommonUtils = {
  println("beginTransaction()…")
  new JdbcCommonUtils
}
​
/**
  * 每接收一条数据后，会调用invoke()方法，将数据写入到MySQL中
  */
override def invoke(transaction: JdbcCommonUtils, value: String, context: SinkFunction.Context): Unit = {
  //将数据写入到MySQL中
  val split = value.split(",")
  val pst = transaction.getConnect.prepareStatement("insert into user(id,name,age) values(?,?,?)")
  pst.setInt(1, Integer.valueOf(split(0)))
  pst.setString(2, split(1))
  pst.setInt(3, Integer.valueOf(split(2)))
​
  //执行插入操作
  pst.execute()
​
  //关闭pst对象
  pst.close()
}
​
/**
  * 当barrier 到达后，Flink会调用preCommit()方法，进行数据预提交
  * 预提交，如果一个preCommit执行失败，其他preCommit也会失败，Flink会调用abort()方法
  */
override def preCommit(transaction: JdbcCommonUtils): Unit = {
  println("barrier 到达，preCommit() 方法执行，开始预提交…")
  //这里的逻辑放在invoke()方法中进行插入数据
}
​
/**
  * Flink checkpoint完成，真正执行提交，Flink在notifyCheckpointComplete()方法中调用该方法，即JobManager完成checkpoint后调用该方法
  */
override def commit(transaction: JdbcCommonUtils): Unit = {
  println("commit() 方法执行…")
  //提交事务
  transaction.commit()
}
​
/**
  * 代码出现异常，事务中止，Flink会调用abort()方法
  * 这里主要是回滚事务
  */
override def abort(transaction: JdbcCommonUtils): Unit = {
  println("abort() 方法执行…")
  //回滚事务
  transaction.rollback()
  //关闭连接
  transaction.close()
}
}

4) 代码测试

在Java或者Scala代码中可以设置并行度为1来观察TwoPhaseCommitSinkFunction抽象类中各个方法执行顺序。启动Java或者Scala代码后，可以观察到commit执行周期与checkpoint周期一致。

向Kafka中输入如下数据：

[root@node1 ~]# kafka-console-producer.sh –bootstrap-server node1:9092,node2:9092,node3:9092 –topic 2pc-topic
1,zs,18
2,ls,19
3,ww,20
4,ml,21
5,tq,22

等待代码中执行commit方法后，查询mysql中数据如下：

mysql> select * from user;
+——+——+——+
| id   | name | age |
+——+——+——+
|   1 | zs   |   18 |
|   2 | ls   |   19 |
|   3 | ww   |   20 |
|   4 | ml   |   21 |
|   5 | tq   |   22 |
+——+——+——+

Flink任务重启与恢复策略

当Flink程序运行失败时，Flink会自动处理任务的重启和恢复，以使任务能够重新回到正常状态。重启策略(Restart Strategies)和故障恢复策略(Failover Strategies)在这个过程中起着关键作用,重启策略决定何时以及是否重新启动失败或受影响的task任务，而故障恢复策略则决定哪些task任务应该被重新启动，以便整个Flink程序能够恢复正常运行状态。

重启策略（Restart Strategies）

默认情况下，Flink使用配置文件"flink-conf.yaml"来设置重启策略，在该文件中，可以通过配置"restart-strategy.type"来选择所需的重启策略。

如果Flink没有启用检查点（checkpoint），默认的重启策略是"no restart"，这意味着任务在失败后不会被重新启动。如果Flink启用了检查点并且没有配置特定的重启策略，那么默认的重启策略是"fixed-delay"策略，即固定延迟启动,重试次数为Integer.MAX_VALUE。

重启策略“restart-strategy.type”参数可以配置以下可用的重启策略：

下面分别介绍以上各种重启策略。

• 无重启策略

Flink 任务失败，不尝试重新启动，如果Flink没有开启checkpoint，默认就是这种重启策略。可以在Flink集群flink-conf.yaml配置文件配置该重启策略：

restart-strategy.type: none

也可以在编写Flink代码时配置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());

• 固定延迟重启策略（fixed delay）

固定延迟重启策略尝试给定次数来重启Flink作业,如果超过最大尝试次数，则Flink作业最终失败，在两次连续的重启尝试之间可以配置等待的间隔时间。如果Flink开启了checkpoint并没有特别指定重启策略，那么默认就是使用“fixed delay”这种重启策略。

可以在Flink集群flink-conf.yaml配置文件中配置该重启策略：

restart-strategy.type: fixed-delay
​
#重启次数,默认1次
restart-strategy.fixed-delay.attempts: 3
​
#重启等待间隔，默认1s
restart-strategy.fixed-delay.delay: 10 s

也可以在编写Flink代码时配置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(  
3, // number of restart attempts  
Time.of(10, TimeUnit.SECONDS) // delay
));

• 故障率重启策略（failure rate）

故障率重启策略当Flink作业失败后重新启动Flink作业，但当超过故障率(每个测量故障次数时间间隔的失败次数)时，作业最终会失败。在两次连续的重启尝试之间可以配置等待的间隔时间。

可以在Flink集群flink-conf.yaml配置文件中配置该重启策略：

restart-strategy.type: failure-rate
#在测量故障率的时间间隔内，Flink作业允许失败最大次数，默认1
restart-strategy.failure-rate.max-failures-per-interval: 3
​
#测量故障率的时间间隔，默认1min，可以设置1 min/20 s格式
restart-strategy.failure-rate.failure-rate-interval: 5 min
​
#任务重试的时间间隔，默认1s，可以设置1 min /20 s格式
restart-strategy.failure-rate.delay: 10 s

也可以在编写Flink代码时配置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.failureRateRestart(  
3, // max failures per interval  
Time.of(5, TimeUnit.MINUTES), //time interval for measuring failure rate  
Time.of(10, TimeUnit.SECONDS) // delay
));

• 指数延迟重启策略（exponential delay）

指数延迟重启策略尝试无限次重启Flink作业，不断增加延迟时间直至最大延迟时间。在两次连续的重启尝试之间，重启策略会呈指数增长，直到达到最大延迟时间并将延迟时间保持在最大值。当作业正确执行时，指数延迟时间值在一段时间阈值后重置,该时间阈值是可配置的。

可以在Flink集群flink-conf.yaml配置文件中配置该重启策略：

restart-strategy.type: exponential-delay
​
#初始延迟，默认1s,可以设置1 min/20 s格式
restart-strategy.exponential-delay.initial-backoff: 10 s
​
#最大延迟，默认5分钟，可以设置1 min/20 s格式
restart-strategy.exponential-delay.max-backoff: 2 min
​
#每次失败后，延迟时间乘以此值作为新的延迟时间，直到达到最大延迟，默认值2.0
restart-strategy.exponential-delay.backoff-multiplier: 2.0
​
#作业运行多长时间后，延迟时间恢复为初始延迟，默认之为1小时
restart-strategy.exponential-delay.reset-backoff-threshold: 10 min
​
#随机偏移值，在延迟基础上随机加上或者减去该值乘以延迟值，防止多个job同时重启，默认值为0.1
restart-strategy.exponential-delay.jitter-factor: 0.1

也可以在编写Flink代码时配置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.exponentialDelayRestart(  
Time.milliseconds(1), // inital backoff  
Time.milliseconds(1000), // max backoff  
1.1, // backoff multiplier  
Time.milliseconds(2000), // threshold duration to reset delay to its initial value  
0.1 // jitter factor
));

注意：以上各种重启策略代码中设置可以覆盖集群flink-conf.yaml文件配置。

故障恢复（Failover Strategies）

Flink支持不同的故障恢复策略，这些策略可以通过flink-conf.yaml文件中参数jobmanager.execution.failover-strategy进行配置。

在“全部重启”故障恢复策略中，Task发生故障时，会重启Flink作业中所有的Task,“局部重启”故障恢复策略中，会将任务分组为不相交的区域，当检测到Task故障时，该策略计算出必须重新启动的最小区域集合从故障中恢复，与“全部重启”策略相比，这种策略重新启动的任务数量较少。