Kafka 流 - 第一个示例 WordCount 不起作用（计数值错误）答案

【问题标题】：Kafka streams - First example WordCount doesn't work (Count values are wrong)Kafka 流 - 第一个示例 WordCount 不起作用（计数值错误）
【发布时间】：2021-04-30 06:21:40
【问题描述】：

我正在使用 KafkaStreams 和 Confluent，并且正在尝试开发“字数”流。

基本上，这个想法是从一个主题中读取（wordcount-input），计算字数，然后将字数写入另一个主题（wordcount-output）。

实现看起来很简单，但是在 Confluent 中查看输出主题时，我没有显示值，而是得到以下输出。

Input topic

Output topic

下面，我附上开发的类型学的代码。

StreamsBuilder builder = new StreamsBuilder();

// reading from input topic
KStream<String, String> textLines = builder.stream("wordcount-input", Consumed.with(Serdes.String(), Serdes.String()));

KStream<String, Long> wordsCount = textLines
   .mapValues((ValueMapper<? super String, ? extends String>) String::toLowerCase)
   .flatMapValues(value -> Arrays.asList(value.split(" "))).selectKey((ignoredKey, word) -> word)
   .groupByKey()
   .count()
   .toStream();
  
// writing to output topic
wordsCount.to("wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

作为测试，我正在通过控制台提取信息，并且工作正常，所以失败是在发送输出主题时。

wordsCount.foreach((k,v) -> System.out.println(k + "--" + v));

你知道发生了什么吗？

附加

// configuration properties
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-test-app");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

【问题讨论】：

标签： spring-boot apache-kafka apache-kafka-streams

【解决方案1】：

我在您的代码片段中发现您的拓扑没有问题。（查看完整代码会很有用）。问题可能出在仪表板中，您应该通过创建 Kafka 消费者来检查，使用 StringDeserializer 作为键，使用 LongDeserializer 作为值，并订阅您的输出主题。不要忘记最早使用 AUTO_OFFSET_RESET_CONFIG 来查看您现有的消息。

【讨论】：