Flink State 误用之痛,竟然 90% 以上的 Flink 开发都不懂
本文主要讨论一个问题:ValueState 中存 Map 与 MapState 有什么区别?
如果不懂这两者的区别,而且使用 ValueState 中存大对象,生产环境很可能会出现以下问题:
- CPU 被打满
- 吞吐上不去
1、 结论
从性能和 TTL 两个维度来描述区别。
性能
- RocksDB 场景,MapState 比 ValueState 中存 Map 性能高很多
- 生产环境强烈推荐使用 MapState,不推荐 ValueState 中存大对象
- ValueState 中存大对象很容易使 CPU 打满
- Heap State 场景,两者性能类似
TTL
Flink 中 State 支持设置 TTL
- MapState 的 TTL 是基于 UK 级别的
- ValueState 的 TTL 是基于整个 key 的
举一反三
能使用 ListState 的场景,不要使用 ValueState 中存 List。
大佬们已经把 MapState 和 ListState 性能都做了很多优化,高性能不香吗?
下文会详细分析 ValueState 和 MapState 底层的实现原理,通过分析原理得出上述结论。
2、 State 中要存储哪些数据
ValueState 会存储 key、namespace、value,缩写为 <K, N, V>。
MapState 会存储 key、namespace、userKey、userValue,缩写为 <K, N, UK, UV>。
解释一下上述这些名词
Key
ValueState 和 MapState 都是 KeyedState,也就是 keyBy 后才能使用 ValueState 和 MapState。所以 State 中肯定要保存 key。
例如:按照 app 进行 keyBy,总共有两个 app,分别是:app1 和 app2。那么状态存储引擎中肯定要存储 app1 或 app2,用于区分当前的状态数据到底是 app1 的还是 app2 的。
这里的 app1、app2 也就是所说的 key。
Namespace
Namespace 用于区分窗口。
假设需要统计 app1 和 app2 每个小时的 pv 指标,则需要使用小时级别的窗口。状态引擎为了区分 app1 在 7 点和 8 点的 pv 值,就必须新增一个维度用来标识窗口。
Flink 用 Namespace 来标识窗口,这样就可以在状态引擎中区分出 app1 在 7 点和 8 点的状态信息。
Value、UserKey、UserValue
ValueState 中存储具体的状态值。也就是上述例子中对应的 pv 值。
MapState 类似于 Map 集合,存储的是一个个 KV 键值对。为了与 keyBy 的 key 进行区分,所以 Flink 中把 MapState 的 key、value 分别叫 UserKey、UserValue。
下面讲述状态引擎是如何存储这些数据的。
3、 StateBackend 中是如何存储和读写 State 数据的
Flink 支持三种 StateBackend,分别是:MemoryStateBackend、FsStateBackend 和 RocksDBStateBackend。
其中 MemoryStateBackend、FsStateBackend 两种 StateBackend 在任务运行期间都会将 State 存储在内存中,两者在 Checkpoint 时将快照存储的位置不同。RocksDBStateBackend 在任务运行期间将 State 存储在本地的 RocksDB 数据库中。
所以下文将 MemoryStateBackend、FsStateBackend 统称为 heap 模式,RocksDBStateBackend 称为 RocksDB 模式。
3.1 Heap 模式 ValueState 和 MapState 是如何存储的
Heap 模式表示所有的状态数据都存储在 TM 的堆内存中,所有的状态都存储的原始对象,不会做序列化和反序列化。(注:Checkpoint 的时候会涉及到序列化和反序列化,数据的正常读写并不会涉及,所以这里先不讨论。)
Heap 模式下,无论是 ValueState 还是 MapState 都存储在 CopyOnWriteStateMap<K, N, V>
中。
key 、 Namespace 分别对应 CopyOnWriteStateMap 的 K、N。
ValueState 的 value 对应 CopyOnWriteStateMap 的 V。
MapState 将会把整个 Map 作为 CopyOnWriteStateMap 的 V,相当于 Flink 引擎创建了一个 HashMap 用于存储 MapState 的 KV 键值对。
具体 CopyOnWriteStateMap 是如何实现的,可以参考《》。
回到正题:Heap 模式下,ValueState 中存 Map 与 MapState 有什么区别?
heap 模式下没有区别。
ValueState 中存 Map,相当于用户手动创建了一个 HashMap 当做 V 放到了状态引擎中。
而 MapState 是 Flink 引擎帮用户创建了一个 HashMap 当做 V 放到了状态引擎中。
所以实质上 ValueState 中存 Map 与 MapState 都是一样的,存储结构都是 CopyOnWriteStateMap<K, N, HashMap>
。区别在于 ValueState 是用户手动创建 HashMap,MapState 是 Flink 引擎创建 HashMap。
3.2 RocksDB 模式 ValueState 和 MapState 是如何存储的
RocksDB 模式表示所有的状态数据存储在 TM 本地的 RocksDB 数据库中。RocksDB 是一个 KV 数据库,且所有的 key 和 value 都是 byte 数组。所以无论是 ValueState 还是 MapState,存储到 RocksDB 中都必须将对象序列化成二进制当前 kv 存储在 RocksDB 中。
3.2.1 ValueState 如何映射成 RocksDB 的 kv
ValueState 有 key、namespace、value 需要存储,所以最简单的思路:
- 将 ValueState 的 key 序列化成 byte 数组
- 将 ValueState 的 namespace 序列化成 byte 数组
- 将两个 byte 数组拼接起来做为 RocksDB 的 key
- 将 ValueState 的 value 序列化成 byte 数组做为 RocksDB 的 value
然后就可以写入到 RocksDB 中。
查询数据也用相同的逻辑:将 key 和 namespace 序列化后拼接起来作为 RocksDB 的 key,去 RocksDB 中进行查询,查询到的 byte 数组进行反序列化就得到了 ValueState 的 value。
这就是 RocksDB 模式下,ValueState 的读写流程。
3.2.2 MapState 如何映射成 RocksDB 的 kv
MapState 有 key、namespace、userKey、userValue 需要存储,所以最简单的思路:
- 将 MapState 的 key 序列化成 byte 数组
- 将 MapState 的 namespace 序列化成 byte 数组
- 将 MapState 的 userKey 序列化成 byte 数组
- 将三个 byte 数组拼接起来做为 RocksDB 的 key
- 将 MapState 的 value 序列化成 byte 数组做为 RocksDB 的 value
然后就可以写入到 RocksDB 中。
查询数据也用相同的逻辑:将 key、namespace、userKey 序列化后拼接起来作为 RocksDB 的 key,去 RocksDB 中进行查询,查询到的 byte 数组进行反序列化就得到了 MapState 的 userValue。
这就是 RocksDB 模式下,MapState 的读写流程。
3.3 RocksDB 模式下,ValueState 中存 Map 与 MapState 有什么区别?
3.3.1 假设 Map 集合有 100 个 KV 键值对,具体两种方案会如何存储数据?
ValueState 中存 Map,Flink 引擎会把整个 Map 当做一个大 Value,存储在 RocksDB 中。对应到 RocksDB 中,100 个 KV 键值对的 Map 集合会序列化成一个 byte 数组当做 RocksDB 的 value,存储在 RocksDB 的 1 行数据中。
MapState 会根据 userKey,将 100 个 KV 键值对分别存储在 RocksDB 的 100 行中。
3.3.2 修改 Map 中的一个 KV 键值对的流程
ValueState 的情况,虽然要修改 Map 中的一个 KV 键值对,但需要将整个 Map 集合从 RocksDB 中读出来。具体流程如下:
- 将 key、namespace 序列化成 byte 数组,生成 RocksDB 的 key
- 从 RocksDB 读出 key 对应 value 的 byte 数组
- 将 byte 数组反序列化成整个 Map
- 堆内存中修改 Map 集合
- 将 Map 集合写入到 RocksDB 中,需要将整个 Map 集合序列化成 byte 数组,再写入
MapState 的情况,要修改 Map 中的一个 KV 键值对,根据 key、namespace、userKey 即可定位到要修改的那一个 KV 键值对。具体流程如下:
- 将 key、namespace、userKey 序列化成 byte 数组,生成 RocksDB 的 key
- 从 RocksDB 读出 key 对应 value 的 byte 数组
- 将 byte 数组反序列化成 userValue
- 堆内存中修改 userValue 的值
- 将 userKey、userValue 写入到 RocksDB 中,需要先序列化,再写入
3.3.3 结论
要修改 Map 中的一个 KV 键值对:
如果使用 ValueState 中存 Map,则每次修改操作需要序列化反序列化整个 Map 集合,每次序列化反序列大对象会非常耗 CPU,很容易将 CPU 打满。
如果使用 MapState,每次修改操作只需要序列化反序列化 userKey 那一个 KV 键值对的数据,效率较高。
举一反三:其他使用 ValueState、value 是大对象且 value 频繁更新的场景,都容易将 CPU 打满。
例如:ValueState 中存储的位图,如果每条数据都需要更新位图,则可能导致 CPU 被打满。
为了便于理解,上述忽略了一些实现细节,下面补充一下:
3.4 直接拼接 key 和 namespace 可能导致 RocksDB 的 key 冲突
假设 ValueState 中有两个数据:
key1 序列化后的二进制为 0x112233, namespace1 序列化后的二进制为0x4455
key2 序列化后的二进制为 0x1122, namespace2 序列化后的二进制为0x334455
这两个数据对应的 RocksDB key 都是 0x1122334455,这样的话,两个不同的 key、namespace 映射到 RocksDB 中变成了相同的数据,无法做区分。
解决方案:
在 key 和 namespace 中间写入 key 的 byte 数组长度,在 namespace 后写入 namespace 的 byte 长度。
写入这两个长度就不可能出现 key 冲突了,具体为什么,读者可以自行思考。
3.5 RocksDB 的 key 中还会存储 KeyGroupId
对 KeyGroup 不了解的同学可以参考《》,加上 KeyGroupId 也比较简单。只需要修改 RocksDB key 的拼接方式,在序列化 key 和 namespace 之前,先序列化 KeyGroupId 即可。
4. State TTL 简述
Flink 中 TTL 的实现,都是将用户的 value 封装了一层,具体参考下面的 TtlValue 类:
public class TtlValue<T> implements Serializable {
@Nullable
private final T userValue;
private final long lastAccessTimestamp;
}
TtlValue 类中有两个字段,封装了用户的 value 且有一个时间戳字段,这个时间戳记录了这条数据写入的时间。
如果开启了 TTL,则状态中存储的 value 就是 TtlValue 对象。时间戳字段也会保存到状态引擎中,之后查询数据时,就可以通过该时间戳判断数据是否过期。
ValueState 将 value 封装为 TtlValue。
MapState 将 userValue 封装成 TtlValue。
ListState 将 element 封装成 TtlValue。
ValueState 中存 Map 与 MapState 有什么区别?
如果 ValueState 中存 Map,则整个 Map 被当做 value,只维护一个时间戳。所以要么整个 Map 过期,要么都不过期。
MapState 中如果存储了 100 个 KV 键值对,则 100 个 KV 键值对都会存储各自的时间戳。因此每个 KV 键值对的 TTL 是相互独立的。
5. 总结
本文从实现原理详细分析了 ValueState 中存 Map 与 MapState 有什么区别?
从性能和 TTL 两个维度来描述两者的区别。
性能
- RocksDB 场景,MapState 比 ValueState 中存 Map 性能高很多,ValueState 中存大对象很容易使 CPU 打满
- Heap State 场景,两者性能类似
TTL
Flink 中 State 支持设置 TTL,TTL 只是将时间戳与 userValue 封装起来。
- MapState 的 TTL 是基于 UK 级别的
- ValueState 的 TTL 是基于整个 key 的
扩展:其实 ListState 的数据映射到 RocksDB 比较复杂,用到了 RocksDB 的 merge 特性,比较有意思,有兴趣的同学可以阅读 RocksDB wiki《Merge Operator Implementation》,链接:https://github.com/facebook/rocksdb/wiki/Merge-Operator-Implementation