Kafka Streams 持久存储清理

Question

是否需要进行一些明确的清理来防止每个持久存储的大小增长太多？我目前正在使用它来计算 DSL API 中的聚合。

Answer 1

我们遇到了类似的问题，我们只是在处理器/变压器中安排了一项清洁存储的工作。只需实现您的 isDataOld(nextValue) 就可以了。

@Override
public void init(ProcessorContext context) {
this.kvStore = (KeyValueStore<Key, Value>) this.context.getStateStore("KV_STORE_NAME");
this.context.schedule(60000, PunctuationType.STREAM_TIME, (timestamp) -> {
    KeyValueIterator<Key, Value> iterator = kvStore.all();
    while (iterator.hasNext()){
    KeyValue<Key,Value> nextValue = iterator.next();
    if isDataOld(nextValue)
       kvStore.delete(nextValue.key);
    }

});
}