java - Kafka Streams KeyValueStore 保留.bytes
问题描述
我对 KeyValueStore 有一个有趣的行为,我有一些假设来解释它,也许你可以告诉我是对还是错......
我配置了一个像下面这样的状态存储
Map<String, String> storeConfig = new HashMap<>();
storeConfig.put(TopicConfig.RETENTION_MS_CONFIG, TimeUnit.DAYS.toMillis(30));
storeConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
StoreBuilder store1 = Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("STORE1"),
Serdes.String(),
Serdes.String()
);
streamsBuilder.addStateStore(store1.withLoggingEnabled(storeConfig));
使用这种配置,我预计 30 天之前的数据集会消失,但我观察到的东西完全不同。
当我查看商店的 rockdb 目录时,它每 14451 个字节滚动文件,并且我在目录中有这样的结构
14451 1. Oct 19:00 LOG
14181 30. Sep 15:59 LOG.old.1569854012833395
14451 30. Sep 17:40 LOG.old.1569918431235734
14451 1. Oct 11:05 LOG.old.1569949239434224
似乎不是在配置的 30 天内实现保留,而是在文件大小上实现。
我在网上发现还有参数Topic.RETENTION_BYTES_CONFIG 'retention.bytes',是不是也要配置这个参数,所以我的数据在保留期间是可见的,不会因为文件大小而被删除(我知道我有我的密钥的值,但在这种现象发生后我无法访问它)...
谢谢回答..
解决方案
在内部,KeyValueStores
使用 RocksDB,而 RocksDB 在内部使用所谓的 LSM-Tree(Log-Structured-Merged-Tree),它会创建许多较小的段,这些段稍后会组合成更大的段。在这个“压缩”步骤之后,可以删除较小的段文件,因为数据被复制到更大的段文件中。因此,没有什么可担心的。
此外,Topic.RETENTION_MS_CONFIG
它是一个主题配置,与 Kafka Streams 应用程序的本地存储无关。此外,aKeyValueStore
将永远保留数据,直到通过“墓碑”消息明确删除。因此,如果您为基础更改日志主题设置保留时间,则可能会在主题中删除数据,但不会在本地存储中删除。
如果要将保留时间应用于本地存储,则不能使用KevValueStore
,但可以使用WindowedStore
它支持保留时间。