Kafka Streams Processor API：根据key将record转发给具体的stream task

Question

我正在使用以下处理器 API 拓扑结构：

Source: STopics (topics: [A, B, C])
  --> P1
Processor: P1 (stores: [P1_Store])
  --> CSink
  <-- STopics
Sink: CSink (topic: Result)
  <-- P1

主题有整数键，比如 ID。每个主题 A、B 和 C 具有相同数量的分区，比如 N 。我有一个场景，其中具有相同键的记录可以出现在任何源主题中。假设我们有 N 个由 P1 处理器（或 N 流任务）创建的实例，每个实例都有一个本地副本国有商店 P1_Store。

有没有办法确定 P1 在运行时的实例数，以便我可以将具有相同密钥的记录发送到处理器的同一实例比如使用 ID % N 来利用本地状态存储中的先前值作为 ID 键？

更新。 我正在重新键入主题 A 中所有具有空键的记录，并且新键 ID 可以出现在任何主题中，这里是 A、B 或 C .

后续问题：如果我有一个处理器P1的两个父处理器P00和P01。 P00 在实例 X 上运行ning 将密钥更改为 123 并转发，而 P01 在实例 Y 上运行ning 并转发密钥 123。如何保证来自 P00-X 和 P01-Y 的密钥 123 始终转到 P1 的相同实例（比如 P1-Z，以便 123 的本地状态存储在 Z 始终可用）？我不希望 P00 和 P01 首先写入中间主题，然后 P1 从该中间主题读取。有没有实现单一拓扑设计的替代方案？

Answer 1

如果你输入的topic已经被key分区了（如果ID设置为message key，默认是什么），你不需要做任何事情。 Kafka Streams 会将分区分配给任务，以便保留分区。

特别是，如果每个主题有 N 个分区，则将有 N 个任务，任务 0 将分配分区 A-0、B-0 和 C-0，依此类推（即具有来自不同主题的相同数字会自动合并）。此外，处理器的实例数与任务数相同。处理器形成任务 X，将处理分区 A-X、B-X 和 C-X 中的所有记录。

如果ID不是你输入的topic中的key，你需要通过一个additional topic设置ID为key后重新分区数据：

// using the DSL
stream.selectKey(...)
      .groupByKey()
      .aggregate(...)

// using Processor API
topology.addSource(...); // read input topics
topology.addProcessor(...); // set ID as key
topology.addSink(...); write to new topic for repartitioning
topology.addSource(...); // read from repartition topic
topology.addProcessor(...); // your processor updating the state

Kafka Streams Processor API：根据key将record转发给具体的stream task

Kafka Streams Processor API: Forward the record to a specific stream task based on the key

apache-kafka

apache-kafka-streams