Apache Flink 进阶(三):Checkpoint 原理解析与应用实践

网友投稿 795 2022-05-29

简介: 大家好,今天我将跟大家分享一下 Flink 里面的 Checkpoint,共分为四个部分。

Checkpoint 是从 source 触发到下游所有节点完成的一次全局操作。下图可以有一个对 Checkpoint 的直观感受,红框里面可以看到一共触发了 569K 次 Checkpoint,然后全部都成功完成,没有 fail 的。

image.png

state 其实就是 Checkpoint 所做的主要持久化备份的主要数据,看下图的具体数据统计,其 state 也就 9kb 大小

image.png

我们接下来看什么是 state。先看一个非常经典的 word count 代码,这段代码会去监控本地的 9000 端口的数据并对网络端口输入进行词频统计,我们本地行动 netcat,然后在终端输入 hello world,执行程序会输出什么?

image.png

答案很明显,(hello, 1) 和 (word,1)

那么问题来了,如果再次在终端输入 hello world,程序会输入什么?

答案其实也很明显,(hello, 2) 和 (world, 2)。为什么 Flink 知道之前已经处理过一次 hello world,这就是 state 发挥作用了,这里是被称为 keyed state 存储了之前需要统计的数据,所以帮助 Flink 知道 hello 和 world 分别出现过一次。

回顾一下刚才这段 word count 代码。keyby 接口的调用会创建 keyed stream 对 key 进行划分,这是使用 keyed state 的前提。在此之后,sum 方法会调用内置的 StreamGroupedReduce 实现。

image.png

Apache Flink 进阶(三):Checkpoint 原理解析与应用实践

又称为 non-keyed state,每一个 operator state 都仅与一个 operator 的实例绑定。

常见的 operator state 是 source state,例如记录当前 source 的 offset

再看一段使用 operator state 的 word count 代码:

image.png

这里的fromElements会调用FromElementsFunction的类,其中就使用了类型为 list state 的 operator state。根据 state 类型做一个分类如下图:

image.png

Managed State:由 Flink 管理的 state,刚才举例的所有 state 均是 managed state

Raw State:Flink 仅提供 stream 可以进行存储数据,对 Flink 而言 raw state 只是一些 bytes

在实际生产中,都只推荐使用 managed state,本文将围绕该话题进行讨论。

下图就前文 word count 的 sum 所使用的StreamGroupedReduce类为例讲解了如何在代码中使用 keyed state:

image.png

下图则对 word count 示例中的FromElementsFunction类进行详解并分享如何在代码中使用 operator state:

image.png

Statebackend 的分类

下图阐释了目前 Flink 内置的三类 state backend,其中MemoryStateBackend和FsStateBackend在运行时都是存储在 java heap 中的,只有在执行 Checkpoint 时,FsStateBackend才会将数据以文件格式持久化到远程存储上。而RocksDBStateBackend则借用了 RocksDB(内存磁盘混合的 LSM DB)对 state 进行存储。

image.png

对于HeapKeyedStateBackend,有两种实现:

支持异步 Checkpoint(默认):存储格式 CopyOnWriteStateMap

仅支持同步 Checkpoint:存储格式 NestedStateMap

特别在 MemoryStateBackend 内使用HeapKeyedStateBackend时,Checkpoint 序列化数据阶段默认有最大 5 MB数据的限制

对于RocksDBKeyedStateBackend,每个 state 都存储在一个单独的 column family 内,其中 keyGroup,Key 和 Namespace 进行序列化存储在 DB 作为 key。

image.png

本小节将对 Checkpoint 的执行流程逐步拆解进行讲解,下图左侧是 Checkpoint Coordinator,是整个 Checkpoint 的发起者,中间是由两个 source,一个 sink 组成的 Flink 作业,最右侧的是持久化存储,在大部分用户场景中对应 HDFS。

第一步,Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint;。

image.png

第二步,source 节点向下游广播 barrier,这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心,下游的 task 只有收到所有 input 的 barrier 才会执行相应的 Checkpoint。

第三步,当 task 完成 state 备份后,会将备份数据的地址(state handle)通知给 Checkpoint coordinator。

image.png

最后,当 Checkpoint coordinator 收集齐所有 task 的 state handle,就认为这一次的 Checkpoint 全局完成了,向持久化存储中再备份一个 Checkpoint meta 文件。

image.png

为了实现 EXACTLY ONCE 语义,Flink 通过一个 input buffer 将在对齐阶段收到的数据缓存起来,等对齐完成之后再进行处理。而对于 AT LEAST ONCE 语义,无需缓存收集到的数据,会对后续直接处理,所以导致 restore 时,数据可能会被多次处理。下图是官网文档里面就 Checkpoint align 的示意图:

image.png

需要特别注意的是,Flink 的 Checkpoint 机制只能保证 Flink 的计算过程可以做到 EXACTLY ONCE,端到端的 EXACTLY ONCE 需要 source 和 sink 支持。

作业恢复时,二者均可以使用,主要区别如下:

image.png

Apache Flink

版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:FPGA笔试题解析(一)
下一篇:【华为云-上云之路】【昇腾】ModelArts与Atlas 200 DK云端协同开发Demo——行人检测(提供完整代码)
相关文章