Kafka Stream: How to trigger event based on hopping windows and how to trigger event based on combined set of windows that are part of hopping window

Viewed 489

Here is our topology:

test-events--->groupByKey-->Window Aggregation-->Suppress-->Filter-->rekey

Here is the flow

We have 1 min hopping window. With total window size of 5 min. The hopping window is evaluated every five minutes, HOW to get count of every 1 min hops??

Total 5 min window, hop duration 1 min.

1. How to TRIGGER some event based on 1 min HOP

2. How to TRIGGER some event based on TOTAL 5 min WINDOW


   ||-----||----||-------||------||-----|

          ||-----||----||-------||------||-----|

We want to do 2 things.

  1. We want to count total number of events in a hopping window of 1 min. If total number of events are > some threshold then trigger one combined event for that 1 min hopping window.

  2. Then we want to trigger an alert if count of combined event is > some number.

From Kafka window API

Code looks like this:
KStream<String, DataPointUri> KS0 = builder.stream(inputTopicsList,);
KGroupedStream<String, DataPointUri> KS1 = KS0.groupByKey();
Duration windowDuration =getDuration(...)); 
Duration hopDuration =getDuration(env.getProperty(...)));
TimeWindowedKStream<String, DataPointUri> KS2 = KS1.windowedBy(TimeWindows.of(windowDuration).advanceBy(hopDuration).grace(Duration.ofSeconds(10)));
KTable<Windowed<String>, Long> uriCount = KS2.count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>with(Serdes.String(), Serdes.Long()));

Here we are getting count for 5 min window. NOT for hop 1 min window. How to get details of 1 min window?

0 Answers
Related