如何在Kafka Streams中删除/清除状态存储? [英] How to remove/clear state stores in Kafka Streams?

问题描述

我在kafka-streams DSL的末尾有一个自定义的 Transformer 实现，并绑定了一个永久的更改日志 KeyValueStore .

I have a custom Transformer implementation in the end of my kafka-streams DSL, with a persistent changelog KeyValueStore bound to it.

自从几周以来，我一直在存储太多数据.现在，每当我加载应用程序时，它都会占用过多的RAM.

Since few weeks I have been putting way too much data in the store. Now whenever I load the application it eats up way too much RAM.

但是，应用程序本身只是一个原型，所以我不介意完全清除商店.

However, the application itself is just a prototype so I don't mind clearing the store entirely.

我可以重命名 kafka.application.id 和 state-store-name ，但这是一个临时解决方法(并且不会删除相应的数据/主题).

I could rename the kafka.application.id and the state-store-name but that's a temporary workaround(and the corresponding data/topics won't be deleted).

我如何彻底清除它?

推荐答案

Confluent的

Confluent's documentation recommends to either use KafkaStreams.cleanUp(), or manually delete directories at /var/lib/kafka-streams/<application.id> (configuration parameter state.dir).

您还需要使用特殊重置工具- bin/kafka-streams-application-reset :

bin/kafka-streams-application-reset --application-id my-streams-app \
                                  --input-topics my-input-topic \
                                  --intermediate-topics rekeyed-topic

此帖子关于重置状态非常有趣.

This post about resetting the state is very interesting.

这篇关于如何在Kafka Streams中删除/清除状态存储?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！