本文根据 Apache Flink 进阶篇系列直播课程整理而成&＃xff0c;由阿里巴巴高级研发工程师唐云&＃xff08;茶干&＃xff09;分享&＃xff0c;主要讲解 Flink 中 Checkpoint 的应用实践&＃xff0c;包括四个部分&＃xff0c;分别是 Checkpoint 与 state 的关系、什么是 state、如何在 Flink 中使用 state 和 Checkpoint 的执行机制。
Tips&＃xff1a;文末可回顾全部基础篇及进阶篇系列教程。
Checkpoint 是从 source 触发到下游所有节点完成的一次全局操作。下图可以有一个对 Checkpoint 的直观感受&＃xff0c;红框里面可以看到一共触发了 569K 次 Checkpoint&＃xff0c;然后全部都成功完成&＃xff0c;没有 fail 的。
state 其实就是 Checkpoint 所做的主要持久化备份的主要数据&＃xff0c;看下图的具体数据统计&＃xff0c;其 state 也就 9kb 大小 。
我们接下来看什么是 state。先看一个非常经典的 word count 代码&＃xff0c;这段代码会去监控本地的 9000 端口的数据并对网络端口输入进行词频统计&＃xff0c;我们本地行动 netcat&＃xff0c;然后在终端输入 hello world&＃xff0c;执行程序会输出什么&＃xff1f;
答案很明显&＃xff0c;(hello, 1) 和 (word,1)。
那么问题来了&＃xff0c;如果再次在终端输入 hello world&＃xff0c;程序会输入什么&＃xff1f;
答案其实也很明显&＃xff0c;(hello, 2) 和 (world, 2)。为什么 Flink 知道之前已经处理过一次 hello world&＃xff0c;这就是 state 发挥作用了&＃xff0c;这里是被称为 keyed state 存储了之前需要统计的数据&＃xff0c;所以帮助 Flink 知道 hello 和 world 分别出现过一次。
回顾一下刚才这段 word count 代码。keyby 接口的调用会创建 keyed stream 对 key 进行划分&＃xff0c;这是使用 keyed state 的前提。在此之后&＃xff0c;sum 方法会调用内置的 StreamGroupedReduce 实现。
对于 keyed state&＃xff0c;有两个特点&＃xff1a;
只能应用于 KeyedStream 的函数与操作中&＃xff0c;例如 Keyed UDF, window state
keyed state 是已经分区 / 划分好的&＃xff0c;每一个 key 只能属于某一个 keyed state
对于如何理解已经分区的概念&＃xff0c;我们需要看一下 keyby 的语义&＃xff0c;大家可以看到下图左边有三个并发&＃xff0c;右边也是三个并发&＃xff0c;左边的词进来之后&＃xff0c;通过 keyby 会进行相应的分发。例如对于 hello word&＃xff0c;hello 这个词通过 hash 运算永远只会到右下方并发的 task 上面去。
再看一段使用 operator state 的 word count 代码&＃xff1a;
这里的 fromElements 会调用 FromElementsFunction 的类&＃xff0c;其中就使用了类型为 list state 的 operator state。根据 state 类型做一个分类如下图&＃xff1a;
除了从这种分类的角度&＃xff0c;还有一种分类的角度是从 Flink 是否直接接管&＃xff1a;
在实际生产中&＃xff0c;都只推荐使用 managed state&＃xff0c;本文将围绕该话题进行讨论。
下图就前文 word count 的 sum 所使用的 StreamGroupedReduce 类为例讲解了如何在代码中使用 keyed state&＃xff1a;
下图则对 word count 示例中的 FromElementsFunction 类进行详解并分享如何在代码中使用 operator state&＃xff1a;
在介绍 Checkpoint 的执行机制前&＃xff0c;我们需要了解一下 state 的存储&＃xff0c;因为 state 是 Checkpoint 进行持久化备份的主要角色。
下图阐释了目前 Flink 内置的三类 state backend&＃xff0c;其中 MemoryStateBackend 和 FsStateBackend 在运行时都是存储在 java heap 中的&＃xff0c;只有在执行 Checkpoint 时&＃xff0c;FsStateBackend 才会将数据以文件格式持久化到远程存储上。而 RocksDBStateBackend 则借用了 RocksDB&＃xff08;内存磁盘混合的 LSM DB&＃xff09;对 state 进行存储。
对于 HeapKeyedStateBackend&＃xff0c;有两种实现&＃xff1a;
特别在 MemoryStateBackend 内使用 HeapKeyedStateBackend 时&＃xff0c;Checkpoint 序列化数据阶段默认有最大 5 MB 数据的限制。
对于 RocksDBKeyedStateBackend&＃xff0c;每个 state 都存储在一个单独的 column family 内&＃xff0c;其中 keyGroup&＃xff0c;Key 和 Namespace 进行序列化存储在 DB 作为 key。
本小节将对 Checkpoint 的执行流程逐步拆解进行讲解&＃xff0c;下图左侧是 Checkpoint Coordinator&＃xff0c;是整个 Checkpoint 的发起者&＃xff0c;中间是由两个 source&＃xff0c;一个 sink 组成的 Flink 作业&＃xff0c;最右侧的是持久化存储&＃xff0c;在大部分用户场景中对应 HDFS。
a. 第一步&＃xff0c;Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint。
b. 第二步&＃xff0c;source 节点向下游广播 barrier&＃xff0c;这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心&＃xff0c;下游的 task 只有收到所有 input 的 barrier 才会执行相应的 Checkpoint。
c. 第三步&＃xff0c;当 task 完成 state 备份后&＃xff0c;会将备份数据的地址&＃xff08;state handle&＃xff09;通知给 Checkpoint coordinator。
d. 第四步&＃xff0c;下游的 sink 节点收集齐上游两个 input 的 barrier 之后&＃xff0c;会执行本地快照&＃xff0c;这里特地展示了 RocksDB incremental Checkpoint 的流程&＃xff0c;首先 RocksDB 会全量刷数据到磁盘上&＃xff08;红色大三角表示&＃xff09;&＃xff0c;然后 Flink 框架会从中选择没有上传的文件进行持久化备份&＃xff08;紫色小三角&＃xff09;。
e. 同样的&＃xff0c;sink 节点在完成自己的 Checkpoint 之后&＃xff0c;会将 state handle 返回通知 Coordinator。
f. 最后&＃xff0c;当 Checkpoint coordinator 收集齐所有 task 的 state handle&＃xff0c;就认为这一次的 Checkpoint 全局完成了&＃xff0c;向持久化存储中再备份一个 Checkpoint meta 文件。
Checkpoint 的 EXACTLY_ONCE 语义
为了实现 EXACTLY ONCE 语义&＃xff0c;Flink 通过一个 input buffer 将在对齐阶段收到的数据缓存起来&＃xff0c;等对齐完成之后再进行处理。而对于 AT LEAST ONCE 语义&＃xff0c;无需缓存收集到的数据&＃xff0c;会对后续直接处理&＃xff0c;所以导致 restore 时&＃xff0c;数据可能会被多次处理。下图是官网文档里面就 Checkpoint align 的示意图&＃xff1a;
需要特别注意的是&＃xff0c;Flink 的 Checkpoint 机制只能保证 Flink 的计算过程可以做到 EXACTLY ONCE&＃xff0c;端到端的 EXACTLY ONCE 需要 source 和 sink 支持。
Savepoint 与 Checkpoint 的区别
作业恢复时&＃xff0c;二者均可以使用&＃xff0c;主要区别如下&＃xff1a;
更多 Apache Flink 基础篇及进阶篇完整教程回顾&＃xff1a;
▼ Flink 社区推荐 ▼
Apache Flink 及大数据领域盛会 Flink Forward Asia 2019 将于 11月28-30日在北京国家会议中心举办&＃xff0c;大会议程已上线&＃xff0c;点击「阅读原文」可了解大会议程详情。
▼
▼
&＃xff08;点击图片可查看 Flink Forward Asia 2019 详情&＃xff09;
你也「在看」吗&＃xff1f;
京公网安备 11010802041100号