How to aggregate over tumbling windows with Kafka Streams

If you have time series events in a Kafka topic, tumbling windows let you group and aggregate them in fixed-size, non-overlapping, contiguous time intervals.

For example, you have a topic with events that represent movie ratings. The following topology definition counts the ratings per title over 10-minute tumbling windows.

  builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), movieRatingSerde))
    .map((key, rating) -> new KeyValue<>(rating.title(), rating))
    .groupByKey(Grouped.with(Serdes.String(), movieRatingSerde))
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1440)))
    .count()
    .toStream()
    .map((Windowed<String> key, Long count) -> new KeyValue<>(key.key(), count))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));

Let's review the key points in this example

    .map((key, rating) -> new KeyValue<>(rating.title(), rating))

Aggregations must group records by key. Since the stream source topic doesn't define any, the code has a map operation which creates new key-value pairs setting the key of the stream to the MovieRating.title field.

    .groupByKey(Grouped.with(Serdes.String(), movieRatingSerde))

Since you've changed the key, under the covers Kafka Streams performs a repartition immediately before it performs the grouping.
Repartitioning is simply producing records to an internal topic and consuming them back into the application. By producing the records the updated keys land on the correct partition. Additionally, since the key-value types have changed you need to provide updated Serde objects, via the Grouped configuration object to Kafka Streams for the (de)serialization process for the repartitioning.

    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1440)))

This creates a new TimeWindowedKStream that we can aggregate. The tumbling windows are 10 minutes long, and we allow data to arrive late by as much as a day.

    .count()

The count() operator is a convenience aggregation method. Under the covers it works like any other aggregation in Kafka Streams i.e. it requires an Initializer, Aggregator and a Materialized to set the Serde for the value since it's a long. But since the result of this aggregation is a simple count, Kafka Streams handles all those details for you.

    .toStream()
    .map((Windowed<String> key, Long count) -> new KeyValue<>(key.key(), count))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));

Aggregations in Kafka Streams return a KTable instance, so it's converted to a KStream. Then map converts to the expected data types.

The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.

Prerequisites

• A Confluent Cloud account
• The Confluent CLI installed on your machine
• Apache Kafka or Confluent Platform (both include the Kafka Streams application reset tool)
• Clone the confluentinc/tutorials repository and navigate into its top-level directory:

  git clone git@github.com:confluentinc/tutorials.git
  cd tutorials

Create Confluent Cloud resources

Login to your Confluent Cloud account:

confluent login --prompt --save

Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:

confluent plugin install confluent-quickstart

Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial. Note that you may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.

confluent quickstart \
  --environment-name kafka-streams-tumbling-windows-env \
  --kafka-cluster-name kafka-streams-tumbling-windows-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./tumbling-windows/kstreams/src/main/resources/cloud.properties

The plugin should complete in under a minute.

Create topics

Create the input and output topics for the application:

confluent kafka topic create ratings
confluent kafka topic create rating-counts

Start a console producer:

confluent kafka topic produce ratings --parse-key --delimiter :

Enter a few JSON-formatted movie ratings:

Super Mario Bros.:{"title":"Super Mario Bros.", "release_year":1993, "rating":3.5, "timestamp":"2024-09-25T11:15:00-0000"}
Super Mario Bros.:{"title":"Super Mario Bros.", "release_year":1993, "rating":2.0, "timestamp":"2024-09-25T11:40:00-0000"}
A Walk in the Clouds:{"title":"A Walk in the Clouds", "release_year":1998, "rating":3.6, "timestamp":"2024-09-25T13:00:00-0000"}
A Walk in the Clouds:{"title":"A Walk in the Clouds", "release_year":1998, "rating":7.1, "timestamp":"2024-09-25T13:01:00-0000"}
Die Hard:{"title":"Die Hard", "release_year":1988, "rating":8.2, "timestamp":"2024-09-25T18:00:00-0000"}
Die Hard:{"title":"Die Hard", "release_year":1988, "rating":7.6, "timestamp":"2024-09-25T18:05:00-0000"}
The Big Lebowski:{"title":"The Big Lebowski", "release_year":1998, "rating":8.6, "timestamp":"2024-09-25T19:30:00-0000"}
The Big Lebowski:{"title":"The Big Lebowski", "release_year":1998, "rating":7.0, "timestamp":"2024-09-25T19:35:00-0000"}
Tree of Life:{"title":"Tree of Life", "release_year":2011, "rating":4.9, "timestamp":"2024-09-25T21:00:00-0000"}
Tree of Life:{"title":"Tree of Life", "release_year":2011, "rating":9.9, "timestamp":"2024-09-25T21:11:00-0000"}

Enter Ctrl+C to exit the console producer.

Compile and run the application

Compile the application from the top-level tutorials repository directory:

./gradlew tumbling-windows:kstreams:shadowJar

Navigate into the application's home directory:

cd tumbling-windows/kstreams

Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:

java -cp ./build/libs/tumbling-windows-standalone.jar \
    io.confluent.developer.TumblingWindow \
    ./src/main/resources/cloud.properties

Validate that you see correct rating counts in the rating-counts topic.

confluent kafka topic consume rating-counts -b \
  --print-key --delimiter : --value-format integer

You should see:

Super Mario Bros.:1
Super Mario Bros.:1
A Walk in the Clouds:1
A Walk in the Clouds:2
Die Hard:1
Die Hard:2
The Big Lebowski:1
The Big Lebowski:2
Tree of Life:1
Tree of Life:1

Clean up

When you are finished, delete the kafka-streams-tumbling-windows-env environment by first getting the environment ID of the form env-123456 corresponding to it:

confluent environment list

Delete the environment, including all resources created for this tutorial:

confluent environment delete <ENVIRONMENT ID>

docker compose -f ./docker/docker-compose-kafka.yml up -d

docker exec -it broker /bin/bash

kafka-topics --bootstrap-server localhost:9092 --create --topic ratings
kafka-topics --bootstrap-server localhost:9092 --create --topic rating-counts

kafka-console-producer --bootstrap-server localhost:9092 --topic ratings \
  --property "parse.key=true" --property "key.separator=:"

Super Mario Bros.:{"title":"Super Mario Bros.", "release_year":1993, "rating":3.5, "timestamp":"2024-09-25T11:15:00-0000"}
Super Mario Bros.:{"title":"Super Mario Bros.", "release_year":1993, "rating":2.0, "timestamp":"2024-09-25T11:40:00-0000"}
A Walk in the Clouds:{"title":"A Walk in the Clouds", "release_year":1998, "rating":3.6, "timestamp":"2024-09-25T13:00:00-0000"}
A Walk in the Clouds:{"title":"A Walk in the Clouds", "release_year":1998, "rating":7.1, "timestamp":"2024-09-25T13:01:00-0000"}
Die Hard:{"title":"Die Hard", "release_year":1988, "rating":8.2, "timestamp":"2024-09-25T18:00:00-0000"}
Die Hard:{"title":"Die Hard", "release_year":1988, "rating":7.6, "timestamp":"2024-09-25T18:05:00-0000"}
The Big Lebowski:{"title":"The Big Lebowski", "release_year":1998, "rating":8.6, "timestamp":"2024-09-25T19:30:00-0000"}
The Big Lebowski:{"title":"The Big Lebowski", "release_year":1998, "rating":7.0, "timestamp":"2024-09-25T19:35:00-0000"}
Tree of Life:{"title":"Tree of Life", "release_year":2011, "rating":4.9, "timestamp":"2024-09-25T21:00:00-0000"}
Tree of Life:{"title":"Tree of Life", "release_year":2011, "rating":9.9, "timestamp":"2024-09-25T21:11:00-0000"}

./gradlew tumbling-windows:kstreams:shadowJar

cd tumbling-windows/kstreams

java -cp ./build/libs/tumbling-windows-standalone.jar \
    io.confluent.developer.TumblingWindow \
    ./src/main/resources/local.properties

kafka-console-consumer --bootstrap-server localhost:9092 --topic rating-counts --from-beginning \
  --property "print.key=true" --property "key.separator=:" \
  --property "value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer"

Super Mario Bros.:1
Super Mario Bros.:1
A Walk in the Clouds:1
A Walk in the Clouds:2
Die Hard:1
Die Hard:2
The Big Lebowski:1
The Big Lebowski:2
Tree of Life:1
Tree of Life:1

docker compose -f ./docker/docker-compose-kafka.yml down