Kafka Streams字数统计应用

本文介绍了Kafka Streams字数统计应用的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在使用kafka流式API(Kakfa版本:0.10.2.0)，试图使一个简单的单词计数示例工作: Wordcount应用要点.我正在同时运行生产者和控制台消费者:

I'm playing around with the kafka streaming API (Kakfa version: 0.10.2.0) trying to make a simple wordcount example work: Wordcount App gist. I'm running both producer and console consumer:

./kafka-console-producer.sh -topic input-topic --broker-list localhost:9092

./kafka-console-consumer.sh --topic output-topic --bootstrap-server localhost:9092 --from-beginning

启动应用程序，一切似乎都工作正常，但是当我在控制台生产者中键入一些字符串时，使用者什么也收不到.如果我更改应用程序以对输入执行简单的toUppercase，则消费者可以接收到流(修改为大写):

start the application and everything seems to be working fine but when I type in some strings within the console producer, the consumer receives nothing at all. If I change the app to do a simple toUppercase on the input the consumer receives the stream (modified to upper case) fine:

//The following code works fine: val uppercasedWithMapValues: KStream[String, String] = textLines.mapValues(_.toUpperCase()) uppercasedWithMapValues.to("output-topic")

//The following code works fine: val uppercasedWithMapValues: KStream[String, String] = textLines.mapValues(_.toUpperCase()) uppercasedWithMapValues.to("output-topic")

有人知道为什么我在单词计数示例中什么也没收到吗?我应该在使用者上指定任何序列化器吗?在我的上一个测试中，控制台使用者处理了我通过控制台发送的消息，但未显示它们，请参见下面的输出:

Does anyone know why I'm receiving nothing on the word-count example? Should I specify any serializer on the consumer? In my last test the console consumer processed the messages that I sent through the console but didn't show them, see below the output:

➜ bin ./kafka-console-consumer.sh \ --topic output-topic \ --bootstrap-server localhost:9092 \ --from-beginning [2017-08-02 07:48:20,187]WARN Error while fetching metadata with correlation id 2 : {output-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) [2017-08-02 07:48:20,197] WARN The following subscribed topics are not assigned to any members in the group console-consumer-91651 : [output-topic] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)

^ C共处理了7条消息

^CProcessed a total of 7 messages

推荐答案

KStream之所以有效，是因为它不使用缓存.对于KTable，您必须稍等一下，或者将cache.max.bytes.buffering设置为0(但不能在生产代码中！)

KStream works because it doesn't use caching. For KTable you have to wait a bit, or set cache.max.bytes.buffering to 0 (but not in a production code!)