Flink学习笔记（五）：Flink 四大基石之 State 和 Checkpoint_flink中checkpoint 和offsetstate

1、Flink 的 State 和 Checkpoint

1.1、State

1.1.1、什么是 State

虽然数据流中的许多操作一次只查看一个单独的事件（例如事件分析器），但某些操作会记住跨多个事件的信息（例如窗口运算符）。这些操作称为有状态操作。

需要状态的一些场景：

• 当应用程序搜索某些事件模式时，状态需要存储截止当前时的事件序列；
• 每分钟/每小时/每天聚合事件时，状态需要处理待处理的聚合；
• 通过数据流训练机器学习模型时，状态保存模型参数的当前版本；
• 当需要管理历史数据时，状态允许有效访问过去发生的事件。

Flink 需要了解状态，以便使用 checkpoints 和 savepoints 使其容错。

状态还允许重新扩展 Flink 应用程序，这意味着 Flink 负责在并行实例之间重新分发状态。

可查询状态允许你在运行时从 Flink 外部访问状态。

在使用状态时， Flink 的 state backends 也可能很有用。Flink 提供了不同的 state backends，指定了状态的存储方式和位置。

1.1.2、状态的应用场景

• 去重，就是去掉重复的数据，避免重复计算，一般而言重复数据定义不同，本处仅仅简单的以主键重复就算重复。例如上游的系统数据可能会有重复，落到下游系统时希望把重复的数据都去掉。去重需要先了解哪些数据来过，哪些数据还没有来，也就是把所有的主键都记录下来，当一条数据到来后，能够看到在主键当中是否存在。
• 窗口计算，窗口计算就是在窗口时间内的数据进行计算，state需要记录哪些数据已经进入窗口，但未进行计算，直到计算结束。
• 机器学习/深度学习，如训练的模型以及当前模型的参数也是一种状态，机器学习可能每次都用有一个数据集，需要在数据集上进行学习，对模型进行一个反馈。
• 访问历史数据，state需要记录哪些数据是历史数据，以便方便的进行当前数据与历史数据比对

1.1.3、有状态计算与无状态计算

• 无状态计算：不需要考虑历史数据，相同的输入得到相同的输出就是无状态计算，如 map/flatMap/filter…
• 有状态计算：需要考虑历史数据，相同的输入（不一定）得到不同的输出就是有状态计算，如 sum/reduce…

1.1.4、状态的分类

1.1.4.1、Managed State & Raw State

Managed State：托管状态
Raw State：原始状态

• 从状态管理方式分：Managed State 由 Flink Runtime 管理，自动存储，自动恢复，在内存管理上有优化；而 Raw State 需要用户自己管理，需要自己序列化，Flink 不知道 State 中存入的数据是什么结构，只有用户自己知道，需要最终序列化为可存储的数据结构。
• 从状态数据结构分：Managed State 支持已知的数据结构，如 Value、List、Map 等。而 Raw State只支持字节数组 ，所有状态都要转换为二进制字节数组才可以。
• 从推荐使用场景分：Managed State 大多数情况下均可使用，而 Raw State 是当 Managed State 不够用时，比如需要自定义 Operator 时，才会使用 Raw State。在实际生产中，都只推荐使用ManagedState。

1.1.4.2、Keyed State & Operator State

Managed State 分为两种，Keyed State（键控状态） 和 Operator State（算子状态） (Raw State都是Operator State)。

• Keyed State（键控状态）

  • 和 key 有关的状态类型，KeyedStream 流上的每一个 key，都对应一个 state；
  • 只能应用于 KeyedStream 的函数与操作中；
  • 存储数据结构：ValueState、ListState、MapState、ReducingState 和 AggregatingState 等等

• Operator State（算子状态）

• 又称为 non-keyed state，每一个 operator state 都仅与一个 operator 的实例（1个SubTask任务）绑定；

• 可以用在所有算子上，每个算子子任务或者说每个算子实例共享一个状态，流入这个算子子任务的数据可以访问和更新这个状态；

• 常见的 operator state 是数据源 source state，例如记录当前 source 的 offset；

• 存储数据结构：ListState 或 BroadcastState 等等

KeyedState案例：

// todo: 自定义状态，实现max算子获取最大值，此处KeyedState定义
		SingleOutputStreamOperator<String> statStream = tupleStream
			// 指定城市字段进行分组
			.keyBy(tuple -> tuple.f0)
			// 处理流中每条数据
			.map(new RichMapFunction<Tuple3<String, String, Long>, String>() {

				// todo: 第1步、定义变量，存储每个Key对应值，所有状态State实例化都是RuntimeContext实例化
				private ValueState<Long> maxState = null ;

				// 处理流中每条数据之前，初始化准备工作
				@Override
				public void open(Configuration parameters) throws Exception {
					// todo: 第2步、初始化状态，开始默认值null
					maxState = getRuntimeContext().getState(
						new ValueStateDescriptor<Long>("maxState", Long.class)
					);
				}

				@Override
				public String map(Tuple3<String, String, Long> value) throws Exception {
					// 获取流中数据对应值
					Long currentValue = value.f2;

					// todo: step3、从状态中获取存储key以前值
					Long historyValue = maxState.value();

					// 如果数据为key分组中第一条数据；没有状态，值为null
					if(null == historyValue ||historyValue < currentValue){
						// todo: step4、更新状态值
						maxState.update(currentValue);
					}

					// 返回状态的最大值
					return value.f0 + " -> " + maxState.value();
				}
			});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

1.1.5、State TTL 状态生命周期

Flink State Time-To-Live：状态的存活时间。

• 在开发Flink应用时，对于许多有状态流应用程序的一个常见要求是自动清理应用程序状态，以有效管理状态大小。
• 从 Flink 1.6 版本开始，社区为状态引入了TTL（time-to-live，生存时间）机制，支持Keyed State 的自动过期，有效解决了状态数据在无干预情况下无限增长导致 OOM 的问题。

设置状态 TTL 过期：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
    
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);
1
2
3
4
5
6
7
8
9
10
11
12

TTL 配置有以下几个选项： newBuilder 的第一个参数表示数据的有效期，是必选项。

TTL 的更新策略（默认是 OnCreateAndWrite）：

• StateTtlConfig.UpdateType.OnCreateAndWrite - 仅在创建和写入时更新
• StateTtlConfig.UpdateType.OnReadAndWrite - 读取时也更新

数据在过期但还未被清理时的可见性配置如下（默认为 NeverReturnExpired):

• StateTtlConfig.StateVisibility.NeverReturnExpired - 不返回过期数据
• StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp - 会返回过期但未清理的数据

NeverReturnExpired 情况下，过期数据就像不存在一样，不管是否被物理删除。这对于不能访问过期数据的场景下非常有用，比如敏感数据。 ReturnExpiredIfNotCleanedUp 在数据被物理删除前都会返回。

注意:

• 状态上次的修改时间会和数据一起保存在 state backend 中，因此开启该特性会增加状态数据的存储。 Heap state backend 会额外存储一个包括用户状态以及时间戳的 Java 对象，RocksDB state backend 会在每个状态值（list
  或者 map 的每个元素）序列化后增加 8 个字节。

• 暂时只支持基于 processing time 的 TTL。

• 尝试从 checkpoint/savepoint 进行恢复时，TTL 的状态（是否开启）必须和之前保持一致，否则会遇到 “StateMigrationException”。

• TTL 的配置并不会保存在 checkpoint/savepoint 中，仅对当前 Job 有效。

• 当前开启 TTL 的 map state 仅在用户值序列化器支持 null 的情况下，才支持用户值为 null。如果用户值序列化器不支持 null， 可以用 NullableSerializer 包装一层。

过期数据清理：
默认情况下，过期数据会在读取的时候被删除，例如 ValueState#value，同时会有后台线程定期清理（如果 StateBackend 支持的话）。可以通过 StateTtlConfig 配置关闭后台清理：

import org.apache.flink.api.common.state.StateTtlConfig;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .disableCleanupInBackground()
    .build();
1
2
3
4
5
6

可以按照如下所示配置更细粒度的后台清理策略。当前的实现中 HeapStateBackend 依赖增量数据清理，RocksDBStateBackend 利用压缩过滤器进行后台清理。

1.2、Checkpoint

Checkpoint：也就是检查点，是用来故障恢复的一种机制。 Spark 也有 Checkpoint ，Flink 与 Spark 一样，都是用 Checkpoint 来存储某一时间或者某一段时间的快照（snapshot），用于将任务恢复到指定的状态。

• State：存储的是某一个 Operator 的运行的状态/历史值，是维护在内存 Memory 中。
• Checkpoint：某一时刻，Flink 中所有 Operator 当前 State 的全局快照，一般存在磁盘上。

Flink 的 Checkpoint 的核心算法叫做 Chandy-Lamport ，是一种分布式快照（Distributed Snapshot）算法，应用到流式系统中就是确定一个 Global 的 Snapshot，错误处理的时候各个节点根据上一次的 Global Snapshot 来恢复。

Checkpoint 实现的核心就是 barrier（栅栏或屏障），Flink 通过在数据集上间隔性的生成屏障 barrier，并通过 barrier 将某段时间内的状态 State 数据保存到 Checkpoint 中（先快照，再保存）。

1. Flink 的 JobManager 创建 Checkpoint Coordinator；
2. Coordinator 向所有的 Source Operator 发送 Barrier 栅栏(理解为执行 Checkpoint 的信号)；
3. Source Operator接收到 Barrier 之后，暂停当前的操作(暂停的时间很短，因为后续的写快照是异步的)，并制作 State 快照, 然后将自己的快照保存到指定的介质中(如 HDFS ), 一切 ok 之后向 Coordinator 汇报并将 Barrier 发送给下游的其他 Operator ；
4. 其他的如 Transformation Operator 接收到 Barrier ，重复第3步，最后将 Barrier 发送给 Sink；
5. Sink接收到Barrier之后重复第3步；
6. Coordinator接收到所有的Operator的执行ok的汇报结果，认为本次快照执行成功；

栅栏对齐：下游 SubTask 必须接收到上游的所有 SubTask 发送 Barrier 栅栏信号，才开始进行 Checkpoint 操作。

1.3、StateBackend 状态后端

Checkpoint 其实就是 Flink 中某一时刻，所有的 Operator 的全局快照，那么快照应该要有一个地方进行存储，而这个存储的地方叫做状态后端（StateBackend）。

• MemoryStateBackend
  • State 存储：TaskManager 内存中
  • Checkpoint 存储：JobManager 内存中
• FsStateBackend
  • State 存储：TaskManager 内存中
  • Checkpoint 存储：可靠外部存储文件系统，本地测试可以为 LocalFS ，测试生产 HDFS。
    • 如果使用 HDFS，则初始化 FsStateBackend 时，需要以 hdfs:// 开头的路径，例如：new FsStateBackend(“hdfs://myhdfs/flink-checkpoint”)
    • 如果使用本地文件，则需要传入以 file:// 开头的路径，录入：new FsStateBackend(“file:///D:/flink-checkpoint”)
• RocksDBStateBackend
  • RocksDB 是一个 嵌入式本地key/value 内存数据库，和其他的 key/value 一样，先将状态放到内存中，如果内存快满时，则写入到磁盘中。类似Redis内存数据库
  • State存储：TaskManager内存数据库（RocksDB）
  • Checkpoint存储：外部文件系统，比如HDFS可靠文件系统中

Flink 1.13 中将状态State和检查点Checkpoint两者区分开来。State Backend 的概念变窄，只描述状态访问和存储；Checkpoint storage，描述的是 Checkpoint 行为，如 Checkpoint 数据是发回给 JM 内存还是上传到远程。

    private static void setEnvCheckpoint(StreamExecutionEnvironment env) {
        // 1.设置checkpoint时间间隔
        env.enableCheckpointing(1000);
        // 2.设置状态后端
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/flink-checkpoints/");
        // 3.没置两个checkpoint 之问最少等待时间，
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // 4.没置checkpoint时失败次数，允许失败几次
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        // 5.设置是否清理检查点，表示 Cancel 时是否需要保当前的
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 6，设置checkpoint的执行供式为EXACTLY_ONCE(默认)，注意: 需要外部支持，如Source和sink的支持
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 7，设置checkpoint的超时时间,如果 Checkpoint在 60s内尚未完成说明该次Checkpoint失败,则丢弃.
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 8。设置同一时间有多少个checkpoint可以同时执行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 9.没置重启策略: NoRestart
        env.setRestartStrategy(RestartStrategies.noRestart());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

程序重启：

• 手动重启
  使用 flink run 运行 Job 执行，指定参数选项 -s path，从Checkpoint检查点启动，恢复以前状态。
• 自动重启

      // 设置自动重后策略，比如最大重启 3 次，每次间隔时间 10 秒
      env.setRestartStrategy( RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
1
2

Savepoint ，手动设置checkpoint：

# Trigger a Savepoint 
$ bin/flink savepoint :jobId [:targetDirectory]

# Trigger a Savepoint with YARN
$ bin/flink savepoint :jobId [:targetDirectory] -yid :yarnAppId

# Stopping a Job with Savepoint
$ bin/flink stop --savepointPath [:targetDirectory] :jobId

# Resuming from Savepoint
$ bin/flink run -s :savepointPath [:runArgs]
1
2
3
4
5
6
7
8
9
10
11

详细区别：

• CheckPoint 的侧重点是“容错”，当Flink作业意外失败，并重启时能直接从早先打下的 CheckPoint 恢复运行，且不影响作业逻辑的准确性。而 SavePoint 侧重点是“维护”，当 Flink 作业需要在人工干预下手动重启、升级、迁移或 A/B 测试时，先将状态整体写入可靠存储，维护完毕之后再从 SavePoint 恢复现场。
• SavePoint 是“通过 CheckPoint 机制”创建的，所以 SavePoint 本质上是特殊的 CheckPoint。
• CheckPoint 面向 Flink Runtime 本身，由 Flink 的各个 TaskManager 定时触发快照并自动清理，一般不需要用户干预；SavePoint 面向用户，完全根据用户的需要触发与清理。
• CheckPoint 的频率往往比较高（因为需要尽可能保证作业恢复的准确度），所以 CheckPoint 的存储格式非常轻量级，但作为 trade-off 牺牲了一切可移植（portable）的东西，比如不保证改变并行度和升级的兼容性。 SavePoint 则以二进制形式存储所有状态数据和元数据，执行起来比较慢而且“贵”，但是能够保证 portability ，如并行度改变或代码升级之后，仍然能正常恢复。
• CheckPoint 是支持增量的（通过RocksDB），特别是对于超大状态的作业而言可以降低写入成本。SavePoint 并不会连续自动触发，所以不支持增量。

1.4、End-to-End Exactly-Once

流处理引擎通常为应用程序提供了三种数据处理语义：最多一次、至少一次和精确一次。
如下是对这些不同处理语义的宽松定义(一致性由弱到强)：

At most noce < At least once < Exactly once < End to End Exactly once
1

• 端到端的精确一次
  结果的正确性贯穿了整个流处理应用的始终，每一个组件都保证了它自己的一致性；Flink 应用从 Source 端开始到 Sink 端结束，数据必须经过的起始点和结束点；

• 实现方式：
  

• 要求：

  • 数据源 Source：支持重设数据的读取位置，比如偏移量 offfset（kafka 消费数据）；
  • 数据转换 Transformation：Checkpoint 检查点机制（采用分布式快照算法实现一致性）；
  • 数据终端 Sink：要么支持幂等性写入，要么事务写入。
    • 幂等写入：
      • Redis 内存 KeyValue 数据库
      • HBase NoSQL数据库
      • MySQL 数据库 replace into 或者 on duplicate key update
    • 事务写入：
      • 预写日志（Write-Ahead-Log）WAL
      • 两阶段提交（Two-Phase-Commit，2PC）

Exactly-once 两阶段提交步骤总结：

• 第1步、Flink 消费到 Kafka 数据之后，就会开启一个 Kafka 的事务，正常写入 Kafka 分区日志但标记为未提交，这就是 Pre-commit（预提交）。

• 第2步、一旦所有的 Operator 完成各自的 Pre-commit ，它们会发起一个 commit 操作。

• 第3步、如果有任意一个 Pre-commit 失败，所有其他的 Pre-commit 必须停止，并且 Flink 会回滚到最近成功完成的 Checkpoint。

• 第4步、当所有的 Operator 完成任务时，Sink 段就收到 Checkpoint barrier（检查点分界线），Sink 保存当前状态存入 Checkpoint ，通知JobManager，并提交外部事务，用于提交外部检查点的数据。

• 第5步、JobManager 收到所有任务的通知，发出确认信息，表示 Checkpoint 已完成，Sink 收到 JobManager 的确认信息，正式 commit （提交）这段时间的数据。

• 第6步、外部系统（Kafka）关闭事务，提交的数据可以正常消费了。

上述过程可以发现，一旦 Pre-commit 完成，必须要确保 commit 也要成功，Operator 和外部系统都需要对此进行保证。整个 两阶段提交协议 2PC 就是解决分布式事务问题，所以才能有如今 Flink 可以端到端精准一次处理。

checkpiont 流程：

1. Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint（触发Checkpoint，发送Barrier栅栏）；
2. 广播barrier并进行持久化;
   1. Source将状态State进行快照，并且进行持久化到存储系统。
   2. Source 节点向下游广播 barrier，这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心，下游的 task 只有收到所有上游的 barrier 才会执行相应的 Checkpoint。
3. 当 task 完成 state 备份后，会将备份数据的地址（state handle）通知给 Checkpointcoordinator;
4. 下游的 sink 节点收集齐上游两个 input 的 barrier 之后（栅栏对齐），将执行本地快照;
   展示了 RocksDB incremental Checkpoint (增量Checkpoint)的流程，首先 RocksDB 会全量刷数据到磁盘上，然后 Flink 框架会从中选择没有上传的文件进行持久化备份。
5. 同样的，sink 节点在完成自己的 Checkpoint 之后，会将 state handle 返回通知Coordinator;
6. 最后，当 Checkpoint coordinator 收集齐所有 task 的 state handle，就认为这一次的Checkpoint 全局完成了，向持久化存储中再备份一个 Checkpoint meta 文件。