{"id":180,"date":"2024-12-12T09:35:58","date_gmt":"2024-12-12T14:35:58","guid":{"rendered":"https:\/\/coding101.xyz\/?p=180"},"modified":"2024-12-12T09:37:22","modified_gmt":"2024-12-12T14:37:22","slug":"kafka-windowed-streams","status":"publish","type":"post","link":"https:\/\/coding101.xyz\/?p=180","title":{"rendered":"Kafka Windowed Streams"},"content":{"rendered":"\n

This is how windowing is done on a Kafka stream – basically calculates maximum of some value in the last five minutes and generates a record with it. <\/p>\n\n\n\n\n\n\n\n

import<\/span> org.apache.kafka.common.serialization.Serdes<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.KafkaStreams<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.StreamsBuilder<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.StreamsConfig<\/span>;<\/span>\nimport<\/span> org.apache.kafka.streams.kstream.*<\/span>;<\/span>\n\nimport<\/span> java.time.Duration<\/span>;<\/span>\nimport<\/span> java.util.Properties<\/span>;<\/span>\n\npublic<\/span> class<\/span> MaxValueLast5Minutes<\/span> {<\/span>\n    public<\/span> static<\/span> void<\/span> main<\/span>(<\/span>String<\/span>[]<\/span> args<\/span>)<\/span> {<\/span>\n        Properties<\/span> props<\/span> =<\/span> new<\/span> Properties<\/span>();<\/span>\n        props<\/span>.<\/span>put<\/span>(<\/span>StreamsConfig<\/span>.<\/span>APPLICATION_ID_CONFIG<\/span>,<\/span> "max-value-5min-app"<\/span>);<\/span>\n        props<\/span>.<\/span>put<\/span>(<\/span>StreamsConfig<\/span>.<\/span>BOOTSTRAP_SERVERS_CONFIG<\/span>,<\/span> "localhost:9092"<\/span>);<\/span>\n        props<\/span>.<\/span>put<\/span>(<\/span>StreamsConfig<\/span>.<\/span>DEFAULT_KEY_SERDE_CLASS_CONFIG<\/span>,<\/span> Serdes<\/span>.<\/span>String<\/span>().<\/span>getClass<\/span>());<\/span>\n        props<\/span>.<\/span>put<\/span>(<\/span>StreamsConfig<\/span>.<\/span>DEFAULT_VALUE_SERDE_CLASS_CONFIG<\/span>,<\/span> Serdes<\/span>.<\/span>Double<\/span>().<\/span>getClass<\/span>());<\/span>\n\n        StreamsBuilder<\/span> builder<\/span> =<\/span> new<\/span> StreamsBuilder<\/span>();<\/span>\n\n        \/\/ Source topic<\/span>\n        KStream<\/span><<\/span>String<\/span>,<\/span> Double<\/span>><\/span> sourceStream<\/span> =<\/span> builder<\/span>.<\/span>stream<\/span>(<\/span>"input-topic"<\/span>);<\/span>\n\n        \/\/ Compute max value for last 5 minutes with a 1-minute hopping window<\/span>\n        KTable<\/span><<\/span>Windowed<\/span><<\/span>String<\/span>>,<\/span> Double<\/span>><\/span> maxValues<\/span> =<\/span> sourceStream<\/span>\n            .<\/span>groupByKey<\/span>()<\/span>\n            .<\/span>windowedBy<\/span>(<\/span>TimeWindows<\/span>.<\/span>ofSizeWithNoGrace<\/span>(<\/span>Duration<\/span>.<\/span>ofMinutes<\/span>(<\/span>5<\/span>))<\/span>\n                                   .<\/span>advanceBy<\/span>(<\/span>Duration<\/span>.<\/span>ofMinutes<\/span>(<\/span>1<\/span>)))<\/span>\n            .<\/span>reduce<\/span>(<\/span>Double:<\/span>:<\/span>max<\/span>);<\/span>\n\n        \/\/ Convert the results to a stream and output<\/span>\n        maxValues<\/span>.<\/span>toStream<\/span>()<\/span>\n                 .<\/span>map<\/span>((<\/span>windowedKey<\/span>,<\/span> value<\/span>)<\/span> -><\/span> KeyValue<\/span>.<\/span>pair<\/span>(<\/span>windowedKey<\/span>.<\/span>key<\/span>(),<\/span> value<\/span>))<\/span>\n                 .<\/span>to<\/span>(<\/span>"output-topic"<\/span>,<\/span> Produced<\/span>.<\/span>with<\/span>(<\/span>Serdes<\/span>.<\/span>String<\/span>(),<\/span> Serdes<\/span>.<\/span>Double<\/span>()));<\/span>\n\n        KafkaStreams<\/span> streams<\/span> =<\/span> new<\/span> KafkaStreams<\/span>(<\/span>builder<\/span>.<\/span>build<\/span>(),<\/span> props<\/span>);<\/span>\n        streams<\/span>.<\/span>start<\/span>();<\/span>\n\n        Runtime<\/span>.<\/span>getRuntime<\/span>().<\/span>addShutdownHook<\/span>(<\/span>new<\/span> Thread<\/span>(<\/span>streams:<\/span>:<\/span>close<\/span>));<\/span>\n    }<\/span>\n}<\/span>\n<\/pre><\/div>\n\n","protected":false},"excerpt":{"rendered":"
This is how windowing is done on a Kafka stream – basically calculates maximum of some value in the last five minutes and generates a record with it.<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-180","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/180","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=180"}],"version-history":[{"count":3,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/180\/revisions"}],"predecessor-version":[{"id":184,"href":"https:\/\/coding101.xyz\/index.php?rest_route=\/wp\/v2\/posts\/180\/revisions\/184"}],"wp:attachment":[{"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=180"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=180"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/coding101.xyz\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=180"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}