apache-kafka - Kafka Streams如何与包含不完整数据的分区一起使用？

Question

Kafka Streams引擎将一个分区恰好映射到一个工作程序(即Java App)，以便该工作程序处理该分区中的所有消息。我有以下情况，并且试图了解它是否仍然可行。

我有一个主题A(带有3个分区)。发送给它的邮件由Kafka随机分区(即没有 key )。我发送给它的消息具有如下所示的架构

{carModel: "Honda", color: "Red", timeStampEpoch: 14334343342}

由于我有3个分区，并且消息在它们之间随机分区，因此可以将相同模型的 汽车写入不同的分区。例如

P1
{carModel: "Honda", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Honda", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Toyota", color: "Blue", timeStampEpoch: 14334343342}

P2
{carModel: "Toyota", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Honda", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Nissan", color: "Blue", timeStampEpoch: 14334343342}

P3
{carModel: "Nissan", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Honda", color: "Red", timeStampEpoch: 14334343342}
{carModel: "Nissan", color: "Blue", timeStampEpoch: 14334343342}

现在，假设我要计算carModel 看到的 的汽车总数。我编写了一个Kafka Streams应用程序，该应用程序监听主题A，并按carModel映射消息，即

carStream.map((key, value) -> KeyValue.pair(value["carModel"], value))

并将总计写到另一个主题B，消息形式为

{carModel: "Nissan", totalCount: 5}

然后，我启动它的3个实例，它们都是的同一消费者组的一部分。然后，Kafka将有效地将每个分区映射到其中一个工作程序。例子

P1 --> Worker A
P2 --> Worker B
P3 --> Worker C

但是，由于每个工作人员仅看到1个分区，因此它将仅看到每个汽车型号的部分信息。它将错过来自其他分区的相同车型的数据。

问题:我的理解正确吗？

如果是这样，我可以想象我可以通过carModel重新分区(即重新排列)我的数据，以使该用例正常工作。

但是我只是想确保我不会误解它是如何工作的，实际上，Kafka确实以某种方式神奇地照顾了我在应用程序中进行内部映射之后的重新分区。

Answer 1

Kafka Streams将自动对您的数据进行重新分区。您的程序将类似于:

stream.map(...).groupByKey().count();

对于这种模式，Kafka Streams检测到您在 map中设置了一个新键，因此将在后台自动创建一个主题，以便为 groupByKey().count()步骤重新分配数据(从v0.10.1开始，通过 KAFKA-3561)。

Note, map() "marks" the stream that it requires repartitioning and .groupByKey().count() will create the topic for repartitioning. With this regard, repartitioning is "lazy", i.e., it is only done if required. If there is no .groupByKey().count() there would be no repartitioning introduced.

基本上，上面的程序的执行方式与

stream.map(...).through("some-topic").groupByKey().count();

Kafka Streams自动“插入” through()步骤，从而计算出正确的结果。

If you are using Kafka Streams 0.10.0, you will need to create the repartition topic manually with the desired number of partitions and you will need to add the call to through() to your code, too.