Flink状态编程:为什么不建议在ValueState里面存Map？
文章目录
先说结论
性能：
● RocksDB 场景，MapState 比 ValueState 中存 Map 性能高很多。
● 生产环境强烈推荐使用 MapState，不推荐 ValueState 中存大对象
● ValueState 中存大对象很容易使 CPU 打满
● Heap State 场景，两者性能类似。
TTL：
MapState的TTL是基于UK的 ( KV 键值对都会存储各自的时间戳)
MapState的TTL是基于key的.
State需要存什么数据
ValueState 会存储 key、namespace、value。
/
* @param <K> The type of the key.
* @param <N> The type of the namespace.
* @param <SV> The type of the values in the state.
*/
public abstract class AbstractHeapState<K, N, SV>
- 1
- 2
- 3
- 4
- 5
- 6
MapState 会存储 key、namespace、userKey、userValue。
/
* @param <K> The type of the key.
* @param <N> The type of the namespace.
* @param <SV> The type of the values in the state.
*/
public abstract class AbstractHeapState<K, N, SV>
- 1
- 2
- 3
- 4
- 5
- 6
● key : 用来keyby的字段
按照 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。
Heap 模式 ValueState 和 MapState 如何存储
Heap表示所有的数据存储在Task Manager的堆内存中，所有状态存为原始对象，不涉及序列化。
Heap模式下，ValueState、MapState都存储在CopyOnWriteStateMap中。
● key、namespace对应CopyOnWriteStateMap的K、N
/*
Type parameters:
<K> – type of key.
<N> – type of namespace.
<S> – type of value.
*/
public class CopyOnWriteStateMap<K, N, S> extends StateMap<K, N, S> {}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
StateBackend模式 如何存储和读写State 数据
1. RocksDB 模式 ValueState 和 MapState 如何存储
RocksDB表示状态数据存储在Task Manager本地的RocksDB数据库中，RocksDB类似Hbase基于LSM树的KV数据库，KV底层存储为Byte数组。
1.1 ValueState如何映射为RocksDB的kv
ValueState的数据结构包含：key、namespace、value
- rocksDb的key：key、namespace序列化byte数组，拼接起来
- rocksDb的key：value序列化byte数组
1.2 MapState如何映射为Rocks DB的kv
MapState的数据结构包含：key、namespace、userKey、userValue
- rocksDb的key：key、namespace、userKey序列化byte数组，拼接起来
- rocksDb的key：userValue序列化byte数组
2. RocksDB下，ValueState存Map 与 MapState有什么区别？
- 假设Map集合有100个KV对，两种方案会如何存储。
-
valueState存Map，那么value=map，rocksdb就会把这100个kv对序列化成一个字节数组存储在rocksdb的一行数据中。
-
mapState:会根据userKey将100个kv对，存在rokcsDB的100行中
2.1 valueState 和 mapState修改流程
● valueState
1、v1、将 key、namespace 序列化成 byte 数组，生成 RocksDB 的 key
2、从 RocksDB 读出 key 对应 value 的 byte 数组
3、将 byte 数组反序列化成整个 Map
4、堆内存中修改 Map 集合
5、将 Map 集合写入到 RocksDB 中，需要将整个 Map 集合序列化成 byte 数组，再写入
● mapState
定位需要修改的key取出对应value，仅修改一行数据，一个vaue即可
- 结论
- 如果使用 ValueState 中存 Map，则每次修改操作需要序列化反序列化整个 Map 集合，
每次序列化反序列大对象会非常耗 CPU，很容易将 CPU 打满。 - 如果使用 MapState，每次修改操作只需要序列化
反序列化 userKey 那一个 KV 键值对的数据，效率较高。
细节补充
valueState直接拼key和namespace可能会导致冲突
假设 ValueState 中有两个数据：
key1 序列化后的二进制为 0x112233， namespace1 序列化后的二进制为0x4455
key2 序列化后的二进制为 0x1122， namespace2 序列化后的二进制为0x334455
这两个数据对应的 RocksDB key 都是 0x1122334455，这样的话，两个不同的 key、namespace 映射到 RocksDB 中变成了相同的数据，无法做区分。
解决方案：
在 key 和 namespace 中间写入 key 的 byte 数组长度，在 namespace 后写入 namespace 的 byte 长度。
写入这两个长度就不可能出现 key 冲突了
RocksDB的key中还会存储KeyGroupId
State TTL
Flink中TTL的实现，都是讲用户的Value封装了一层。源码如下：
public class TtlValue<T> implements Serializable {
private static final long serialVersionUID = 5221129704201125020L;
@Nullable private final T userValue;
private final long lastAccessTimestamp; // 数据写入时间
}
- 1
- 2
- 3
- 4
- 5
- 6
- ValueState 将 value 封装为 TtlValue。
- MapState 将 userValue 封装成 TtlValue。
- ListState 将 element 封装成 TtlValue。
结论：
如果 ValueState 中存 Map，则整个 Map 被当做 value，只维护一个时间戳。所以要么整个 Map 过期，要么都不过期。
MapState 中如果存储了 100 个 KV 键值对，则 100 个 KV 键值对都会存储各自的时间戳。因此每个 KV 键值对的 TTL 是相互独立的。