app建设网站公司简介,先做它个天猫网站,网站用什么语言编写,建设银行衡阳市分行网站flink作业中的MapState开启了TTL#xff0c;并且使用rocksdb作为状态后端配置了全量快照方式#xff08;同时启用全量快照清理#xff09;#xff0c;希望能维持一个平稳的运行状态#xff0c;但是经观察后发现效果不达预期#xff0c;不仅checkpoint size持续缓慢递增并且使用rocksdb作为状态后端配置了全量快照方式同时启用全量快照清理希望能维持一个平稳的运行状态但是经观察后发现效果不达预期不仅checkpoint size持续缓慢递增很长时间后还发生了物理内存溢出。
场景复现 env.addSource(new OneRecordSource(100000000, 1)).keyBy(e - e).map(new RichMapFunctionString, String() {private MapStateString, String map;Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);StateTtlConfig ttlConfig StateTtlConfig.newBuilder(Time.minutes(1)).cleanupFullSnapshot().build();MapStateDescriptorString, String lastStateDes new MapStateDescriptor(lastState, String.class, String.class);lastStateDes.enableTimeToLive(ttlConfig);map getRuntimeContext().getMapState(lastStateDes);}Overridepublic String map(String key) throws Exception {map.put(key, key 值值值值值值值值值值值值值值值值值);return key;}}).print();自定义一个简单的source持续递增生成整数作为key保存到MapState设置了1分钟的过期时间从web中看到即使过了几分钟但是checkpoint size依然是稳步增加的说明了过期的数据并没得到清理。
原因分析
现象中全量checkpoint size增长说明了两个问题
本地rocksdb持续增长过期数据在本地状态得没得到清理远端快照文件持续增长过期数据在快照过程依然被保留了
rocksdb作为状态后端时依赖的是压缩时清理过期数据具有滞后性越久的数据处于更上层压缩频率更小这解释了发生问题1的原因。在启用了全量快照清理条件下就算本地状态依然保留着过期数据在发生全量快照的时候为什么不把过期数据过滤掉造成checkpoint size单调递增理想的情况是随着ttl发生周期性的增减。
在全量快照时在RocksDBMapState.StateSnapshotTransformerWrapper.filterOrTransform中会对本地所有状态数据根据ttl配置进行过滤转换对于已过期的key其value设为NULL_VALUE(长度为1的byte[])这样造成了过期的kv依然保留在远程端只是原始的值使了统一的标识代替但是hashmap的MapState在该环节会把过期的kv直接过滤掉。造成这种处理上的差异不知道是什么出发点不过测试将过期的value修改为null后在一个ttl周期后checkpoint size趋向于固定效果和hashmap一致。
psRocksDbListState效果正常。
总结
rocksdb的压缩与sst文件数量和大小有关所以猜测全量快照大小到达一个比较大的值后应该不会继续增长可以调整rocksdb的压缩策略使压缩变得更“积极”但是肯定会消耗更多资源rocksdb MapState TTL组合使用建议使用增量快照方式
