使用kafka流(版本0.10.0.1)和kafka代理(版本0.10.0.1),我尝试基于消息键生成计数。我使用以下命令生成消息:
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-streams-topic --property parse.key=true --property key.separator=,
当我运行上述命令时,我可以发送如下键和值:
1,{"value":10}
这将向kafka发送一个key=1,value={“value”:10}的消息。
我的目标是计算有多少条消息的键=1。给定上述命令,计数将为1。
下面是我使用的代码:
public class StreamProcessor {
public static void main(String[] args) {
KStreamBuilder builder = new KStreamBuilder();
final Serde<Long> longSerde = Serdes.Long();
final Serde<String> stringSerde = Serdes.String();
KStream<String, String> values = builder.stream(stringSerde, stringSerde, "kafka-streams-topic");
KStream<String, Long> counts = values
.countByKey(stringSerde, "valueCounts")
.toStream();
counts.print(stringSerde, longSerde);
counts.to(stringSerde, longSerde, "message-counts-topic");
KafkaStreams streams = new KafkaStreams(builder, properties());
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
private static Properties properties() {
final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-poc");
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
return streamsConfiguration;
}
}
当我运行counts.print(stringserde,lonserde)时,我得到:
1 , 1
这意味着我有一个key=1,而他们的消息是一个有这个key的消息。这就是我所期望的。
但是,当以下行运行时:
counts.to(stringSerde, longSerde, "message-counts-topic");
名为message counts topic的主题获取发送给它的消息,但当我尝试使用以下命令读取消息时:
./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic message-counts-topic --property print.key=true --property key.separator=, --from-beginning
我得到以下输出:
1 ,
其中1是键,值不显示任何内容。我希望看到信息1,1。但是由于某些原因,即使在调用print方法时显示其值,计数值也会丢失。
1条答案
按热度按时间7cjasjjr1#
您需要为指定一个不同的值反序列化程序
bin/kafka-console-consumer.sh
. 添加以下内容:默认字符串反序列化程序无法正确读取长值。