我可以避免在下面的卡夫卡流中重新分区吗

Question

我正在尝试通过更改键从主题流创建一个 table，但值保持不变。是否可以避免重新分区？

streamsBuilder.stream(
            TOPIC,
            Consumed.with(IdSerde(), ValueSerde())
        )
            .peek { key, value -> logger.info("Consumed $TOPIC, key: $key, value: $value") }
            .filter { _, value -> value != null }
            .selectKey(
                { _, value -> NewKey(value.newKey.toString()) },
                Named.`as`("changeKey")
            )
            .toTable(
                Materialized.`as`<NewKey, Value, KeyValueStore<Bytes, ByteArray>>(
                    NEW_TABLE_NAME
                )
                    .withKeySerde(NewKeySerde())
                    .withValueSerde(ValueSerde())
            )
        return streamsBuilder.build()

Answer 1

我能想到的唯一方法是您的生产者应该使用适当的密钥将数据发送到输入主题，而不是您在流应用程序中选择密钥。

否则，答案是否。

在开始解释之前，先了解一下 KTable

的一些背景知识

A KTable is an abstraction of a changelog stream. This means a KTable holds only the latest value for a given key.

考虑将以下四个记录（按顺序）发送到流

("alice", 1) --> ("bob", 100) --> ("alice", 3) --> ("bob", 50)

对于以上内容，KTable 如下所示（alice 和 bob 是消息的键）：

("alice", 3) ("bob", 50)

说明

Whenever we run multiple instances of the kafka streams application, the KTable in each instance holds only the local state of the application. Hence, repartitioning is required to ensure that the messages with the same key land in the same partition.

为了更好地理解这一点，让我们考虑一个例子。我们有一个包含两个分区的主题。 Partition 0有一个事件，Partition 1有三个事件如下（null表示没有key）：

Topic Partition 0: (null, {"name": alice,"count": 1}) , Topic Partition 1: (null, {"name": alice,"count": 3}) , (null, {"name": "bob","count": 100}), (null, {"name": "bob","count": 50})

我们创建一个kafka streams 应用程序来从这个主题中读取数据，并使用名称字段作为键创建一个KTable。此外，流应用程序是运行两个实例。每个实例将被分配一个单独的分区，如下所示：

Topic Partition 0: -----> Instance 1 Topic Partition 1: -----> Instance 2

由于每个实例都在本地维护 KTable，如果不进行重新分区 - KTable 将处于两个实例的不一致状态。如下所示：

Instance 1 KTable : ("alice", {"count":1}) Instance 2 KTable : ("alice", {"count":1}), ("bob", {"count":50})

因此，如果在 selectKey 操作之后创建 KTable，为避免上述问题，kafka 流会重新分区主题。

我可以避免在下面的卡夫卡流中重新分区吗

Can I avoid repartition in the below kafka stream

apache-kafka

apache-kafka-streams