如何删除/清除 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.

几周以来，我一直在存储中放入太多数据.现在，每当我加载应用程序时，它都会占用太多内存.

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 的 文档建议 要么使用 KafkaStreams.cleanUp()，要么手动删除 /var/lib/kafka-streams/(配置参数state.dir).

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:

You also need to reset all topics used by application with use of special reset tool - 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屋！