我想做的是:使用数字主题(Long‘s)中的记录汇总(计数)每5秒窗口的值将最终聚合结果发送到另一个主题我的代码看起来如下:KStream<String, Long> longs = builder.stream(
Serdes.String(), Serdes.Long(), "longs");
// In one ktable, count by key, on a five second tumbling window.
KTable<Windowed<String>, Long> longCounts =
longs.countByKey(TimeWindows.of("longCounts", 5000L));
// Finally, sink to the long-avgs topic.
longCounts.toStream((wk, v) -> wk.key())
.to("long-counts");看起来一切都像预期的那样工作,但是聚合被发送到每个传入记录的目标主题。我的问题是如何只发送每个窗口的最终聚合结果?
2 回答

慕哥9229398
TA贡献1877条经验 获得超6个赞
没有找到匹配的内容?试试慕课网站内搜索吧
添加回答
举报
0/150
提交
取消