代码之家 › 专栏 › 技术社区 › user1028741

如何仅在窗口结束时输出窗口聚合的结果?[副本]

apache-kafka-streams apache-kafka

user1028741 · 技术社区 · 7 年前

我有一个 KStream 我想在其中计算一些事件的维度。我的做法如下:

KTable<Windowed<Long>, Counter> ret = input.groupByKey()
  .windowedBy(TimeWindows.of(Duration.of(10, SECONDS)))
  .aggregate(Counter::new, (k, v, c) -> new Counter(c.count + v.getDimension()));

我想要一个新的 KStream

ret.toStream().to("output");

问题是“输入”主题中的每个事件都会产生一个事件到“输出”主题。我希望仅在窗口完成时将事件发布到输出主题。例如,如果窗口为一分钟,则每分钟每键发送一个事件。

我想我可以这样做:

ret.toStream().foreach((k, v) -> sendToKafkaTopic("output"));

1 回复 | 直到 7 年前

Stefan Repcek 7 年前

您可以使用KTable的新功能 KTable.suppress

此方法允许您为窗口计算的每个窗口/关键点仅获得一个最终结果。

更多关于 suppres 在里面 KIP-328

您可以使用更新您的实现 suppress 这样地:

KTable<Windowed<Long>, Counter> ret = input.groupByKey()
        .windowedBy(TimeWindows.of(Duration.of(10, SECONDS)))
        .aggregate(Counter::new, (k, v, c) -> new Counter(c.count + v.getDimension()))
        .suppress(untilWindowCloses(BufferConfig.unbounded()));

ret.toStream().to("output"); // now stream should flush events to the output topic only when the window closes

推荐文章

Stella · Kafka流处理期间处理数据时出现异常

8 年前

Stella · kafka流中的处理器节点

8 年前

mukesh210 · 卡夫卡流:Lib Rocks数据库上的UnsatifiedLinkError

8 年前

el323 · Kafka流:通过Json日志中的键进行分组

8 年前

Suchita · 如何使用Avro反序列化器和Avro序列化器为KafkaStreams编写Junit测试用例

8 年前

Cemo · 如何获得有关GlobalKTable状态存储更新的通知?

8 年前

px5x2 · Kafka将GlobalKTable同步流传输到应用程序

8 年前

Dth · 加工机为Kafka流中的单个任务多次调用init()

8 年前

nattyddubbs · Kafka Streams多实例设计含义

8 年前

dvlcis · 如何使用卡夫卡流窗口为烛台图生成创建一条记录

8 年前