Kafka 流如何处理分布式数据

Question

我尝试过各种教程，但不清楚 Kafka 流的两个方面。 让我们以下面提到的字数统计为例： https://docs.confluent.io/current/streams/quickstart.html

// Serializers/deserializers (serde) for String and Long types
final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();

// Construct a `KStream` from the input topic "streams-plaintext-input", where message values
// represent lines of text (for the sake of this example, we ignore whatever may be stored
// in the message keys).
KStream<String, String> textLines = builder.stream("streams-plaintext-input", Consumed.with(stringSerde, stringSerde));

KTable<String, Long> wordCounts = textLines
// Split each text line, by whitespace, into words.  The text lines are the message
// values, i.e. we can ignore whatever data is in the message keys and thus invoke
// `flatMapValues` instead of the more generic `flatMap`.
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
// We use `groupBy` to ensure the words are available as message keys
.groupBy((key, value) -> value)
// Count the occurrences of each word (message key).
.count();

// Convert the `KTable<String, Long>` into a `KStream<String, Long>` and write to the output topic.
wordCounts.toStream().to("streams-wordcount-output", 
Produced.with(stringSerde, longSerde));

这里有几个问题：
1.）由于原始流中没有键，因此两个单词可以落在两个不同的节点上，因为它们可能属于不同的分区，因此真正的计数将是它们两者的聚合。这里好像没有做？服务于同一主题的分区的不同节点是否在此处协调以汇总计数？
2.) 由于每个操作（例如 flatMapValues、groupBy 等）都生成了新流，因此这些子流中的消息是否会重新计算分区，以便它们落在不同的节点上？

将不胜感激这里的任何帮助！

Answer 1

1.) 由于原始流中没有键，因此两个单词可能落在两个不同的节点上，因为它们可能落在不同的分区中，因此真正的计数将是它们两者的聚合。这里好像没有做？

到这里就完成了。这是相关代码：

// We use `groupBy` to ensure the words are available as message keys
.groupBy((key, value) -> value)

这里，“单词”成为新的消息键，这意味着单词被重新分区，每个单词只放入一个分区。

服务于同一主题的分区的不同节点是否在此处协调以汇总计数？

不，他们没有。分区仅由一个节点处理（更准确地说：仅由一个流任务处理，见下文）。

2.) 由于每个操作（例如 flatMapValues、groupBy 等）都会生成新流，因此这些子流中的消息会重新计算分区，以便它们落在不同的节点上？

不确定我是否理解您的问题，尤其是“重新计算”的评论。操作（如聚合）总是按分区执行，Kafka Streams 将分区映射到流任务（稍微简化：一个分区总是由一个且只有一个流任务处理）。流任务由 Kafka Streams 应用程序的各种实例执行，这些实例通常在不同的容器/VM/机器上运行。如果需要，数据将需要重新分区（请参阅上面的问题 #1 和答案），以使操作产生预期的结果——也许这就是您所说的“重新计算”的意思。

我建议阅读 Kafka 的文档，例如 https://kafka.apache.org/documentation/streams/architecture#streams_architecture_tasks。