假设groupby函数在kafka流中不可用。我可以做下面的得到字数和建立一个ktable在上面吗?请注意,我在拓扑中使用了两次“字数主题”。我有一个用例,我想迭代地构建一些东西,对于下一个流事件,我想查找上一个值并基于事件更新它。我想在构建ktable的同一主题中保留最新的值。
KTable<String,Long> wordCountTable = builder.table("word-count-topic",Consumed.with(Serdes.String(), Serdes.Long()));
KStream<String,String> wordsStream = builder.stream("words-topic",Consumed.with(Serdes.String(), Serdes.String()));
KStream<String,String> msgStream = wordsStream
.flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
.selectKey((k,v) -> v);
msgStream.leftJoin(kTable, (word,count) -> {
if( count == null) return new WordCount(word, Long.valueOf(1));
else return new WordCount(word, count + 1);
})
.mapValues((k,v)-> v.getCount())
.to("word-count-topic", Produced.with(Serdes.String(), Serdes.Long()));
streams = new KafkaStreams(builder.build(), props);
streams.start();
1条答案
按热度按时间yacmzcpb1#
这应该管用。为什么不直接运行代码呢?