python怎么开发网站,互联网服务提供商,网站 缓存什么时候使用,360推广登录平台flink的keyed state是有有效期(TTL)的#xff0c;使用和说明在官网描述的篇幅也比较多#xff0c;对于三种清理策略没有进行横向对比得很清晰。
全量快照清理(FULL_STATE_SCAN_SNAPSHOT)增量清理(INCREMENTAL_CLEANUP)rocksdb压缩清理(ROCKSDB_COMPACTION_FILTER) 注意…flink的keyed state是有有效期(TTL)的使用和说明在官网描述的篇幅也比较多对于三种清理策略没有进行横向对比得很清晰。
全量快照清理(FULL_STATE_SCAN_SNAPSHOT)增量清理(INCREMENTAL_CLEANUP)rocksdb压缩清理(ROCKSDB_COMPACTION_FILTER) 注意三种状态清理策略不是互斥的并不是三选一的问题一般是全量快照清理配合另两个其中的一个来使用需要根据不同的state backend可以看到StateTtlConfig.CleanupStrategies.strategies是一个集合来的。
全量快照清理
只发生在全量checkpoint的时候把过期的元素过滤掉以减少远程端checkpoint storage的大小(如hdfs)无论状态后端是hashmap还是rocksdb都支持官网用了另一种相同意思描述在 RocksDBStateBackend 的增量 checkpoint 模式下无效(当前hashmap不支持增量checkpoint)
开启方式 StateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.seconds(1)).cleanupFullSnapshot() //只要有这行即可.build();
// 只有开启了全量快照清理才会触发transform过滤private StateSnapshotTransformFactory? getSnapshotTransformFactory() {if (!ttlConfig.getCleanupStrategies().inFullSnapshot()) {return StateSnapshotTransformFactory.noTransform();} else {return new TtlStateSnapshotTransformer.Factory(timeProvider, ttl);}} 当进行全量快照是会遍历状态中每个元素(TtlValue)过滤掉那些已经过期的TtlStateSnapshotTransformer.expired()
增量清理
在状态访问、增加、修改、删除时都会触发主要是为了减少状态后端中的大小当前仅状态后端为hashmap才支持增量清理
这里的增量有点难理解实际它的含义是对于一个keyed stated状态是按key被分区成若干部分的每次只取cleanupSize个key的状态(StateEntry)进行清理对于大状态这是必须的因为这会增加状态算子处理数据的时间。对于rocksdb由于提供不了全局的StateEntry访问器所以它不支持增量清理AbstractRocksDBState.getStateIncrementalVisitor
判断是否满足增量清理TtlStateFactory.registerTtlIncrementalCleanupCallback 清理逻辑TtlIncrementalCleanup.runCleanup
启用方式这玩意在hashmap状态后端中是默认开启的!!! public IncrementalCleanupStrategy getIncrementalCleanupStrategy() {IncrementalCleanupStrategy defaultStrategy isCleanupInBackground ? DEFAULT_INCREMENTAL_CLEANUP_STRATEGY : null;return (IncrementalCleanupStrategy)strategies.getOrDefault(Strategies.INCREMENTAL_CLEANUP, defaultStrategy);}只要isCleanupInBackgroundtrue就算没显式调用cleanupIncrementally(),它依然会给你生成一个默认的增量清理策略。所以在上述全量快照清理的启用示例中同时也开启了增量清理。如果需要禁止需要显式调disableCleanupInBackground()
rocksdb压缩清理
flink 提供的 rocksdb 压缩过滤器(FlinkCompactionFilter)会在压缩时过滤掉已经过期的状态数据最终调用到C的nactive方法减少sst文件的大小。应该与增量/全量checkpoint方式无关。
