如何发送时间窗口化KTable的最终kafka-streams聚合结果? [英] How to send final kafka-streams aggregation result of a time windowed KTable?
问题描述
我想做的是这样
- 使用数字主题(长号)消费记录
- 汇总(计数)每个5秒窗口的值
- 将FINAL汇总结果发送到另一个主题
我的代码如下:
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");
看起来一切都按预期工作,但是聚合被发送到每个传入记录的目标主题.我的问题是如何只发送每个窗口的最终聚合结果?
It looks like everything works as expected, but the aggregations are sent to the destination topic for each incoming record. My question is how can I send only the final aggregation result of each window?
推荐答案
在Kafka Streams中,没有最终汇总"之类的东西. Windows始终保持打开状态,以处理在窗口结束时间过去之后到达的乱序记录.但是,窗户不会永远保持下去.他们的保留时间到期后将被丢弃.何时丢弃窗口没有特殊动作.
In Kafka Streams there is no such thing as a "final aggregation". Windows are kept open all the time to handle out-of-order records that arrive after the window end-time passed. However, windows are not kept forever. They get discarded once their retention time expires. There is no special action as to when a window gets discarded.
有关更多详细信息,请参见Confluent文档: http://docs.confluent.io/current/streams/
See Confluent documentation for more details: http://docs.confluent.io/current/streams/
因此,对于聚合的每次更新,都会生成一个结果记录(因为Kafka Streams还会在无序记录上更新聚合结果).您的最终结果"将是最新的结果记录(在丢弃窗口之前).根据您的用例,手动重复数据删除将是解决问题的一种方法(使用较低级别的API,transform()
或process()
)
Thus, for each update to an aggregation, a result record is produced (because Kafka Streams also update the aggregation result on out-of-order records). Your "final result" would be the latest result record (before a window gets discarded). Depending on your use case, manual de-duplication would be a way to resolve the issue (using lower lever API, transform()
or process()
)
此博客文章也可能有帮助: https://timothyrenner.github.io/engineering/2016/08/11/kafka-streams-not-looking-at-facebook.html
This blog post might help, too: https://timothyrenner.github.io/engineering/2016/08/11/kafka-streams-not-looking-at-facebook.html
Another blog post addressing this issue without using punctuations: http://blog.inovatrend.com/2018/03/making-of-message-gateway-with-kafka.html
更新
使用添加了KTable#suppress()
运算符KIP-328 ,它可以严格禁止连续更新,并且每个窗口仅发出一个结果记录;权衡是增加等待时间.
With KIP-328, a KTable#suppress()
operator is added, that will allow to suppress consecutive updates in a strict manner and to emit a single result record per window; the tradeoff is an increase latency.
这篇关于如何发送时间窗口化KTable的最终kafka-streams聚合结果?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!