时间戳

首页 > 解决方案 > 使用kafka流在时间窗口中获取给定键的最后一个事件

java - 使用kafka流在时间窗口中获取给定键的最后一个事件

问题描述

我开始使用 KStream 来使用来自现有主题的数据。

我只对在 10 秒窗口内获取给定 ID 的最后一个事件感兴趣。我尝试使用以下代码：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, MySale> stream = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), specificAvroSerde));

stream.selectKey((key, value) -> value.getID())
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
    .reduce((value1, value2) -> value2)
    .toStream()
    .peek((key, value) -> log.info("key={}, value={}", key, value.toString()))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), specificAvroSerde));

但我最终得到了所有的事件，而不仅仅是最后一个。是否可以使用 KStream 做我想做的事？

标签： javaapache-kafkastreamdslapache-kafka-streams

解决方案

采用.suppress()

它抑制了窗口中的所有中间结果，只发出最终结果。

stream.selectKey((key, value) -> value.getID())
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
    .reduce((value1, value2) -> value2)
    .suppress(Suppressed.untilWindowCloses(unbounded())))  // like this
    .toStream()
    .peek((key, value) -> log.info("key={}, value={}", key, value.toString()))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), specificAvroSerde));

你可以在这里阅读更多：https ://kafka.apache.org/21/documentation/streams/developer-guide/dsl-api.html#window-final-results

推荐阅读