An aggregation in Kafka Streams is a stateful operation used to perform a "clustering" or "grouping" of values with the same key. An aggregation in Kafka Streams may return a different type than the input value. In our example here we're going to use the count() method to perform a count on the number of tickets sold.
builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), ticketSaleSerde))
.map((k, v) -> new KeyValue<>(v.title(), v.ticketTotalValue()))
.groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))
.count()
.toStream().mapValues(v -> v.toString() + " tickets sold")
.to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.String()));
Let's review the key points in this example
map((key, value) -> new KeyValue<>(v.title(), v.ticketTotalValue()))
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 TicketSale.title field.
groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))
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.
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().mapValues(v -> v + " tickets sold")
.to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.String()));
Aggregations in Kafka Streams return a KTable instance, so it's converted to a KStream then mapValues appends a string to the count to give it some context on the meaning of the number.
The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.
git clone git@github.com:confluentinc/tutorials.git
cd tutorials
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-aggregating-sum-env \
--kafka-cluster-name kafka-streams-aggregating-sum-cluster \
--create-kafka-key \
--kafka-java-properties-file ./aggregating-sum/kstreams/src/main/resources/cloud.properties
The plugin should complete in under a minute.
Create the input and output topics for the application:
confluent kafka topic create aggregation-sum-input
confluent kafka topic create aggregation-sum-output
Start a console producer:
confluent kafka topic produce aggregation-sum-input
Enter a few JSON-formatted ticket sales:
{"title":"Guardians of the Galaxy", "ticketTotalValue":15}
{"title":"Doctor Strange", "ticketTotalValue":15}
{"title":"Guardians of the Galaxy", "ticketTotalValue":15}
Enter Ctrl+C to exit the console producer.
Compile the application from the top-level tutorials repository directory:
./gradlew aggregating-sum:kstreams:shadowJar
Navigate into the application's home directory:
cd aggregating-sum/kstreams
Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:
java -cp ./build/libs/aggregating-sum-standalone.jar \
io.confluent.developer.KafkaStreamsAggregatingCount \
./src/main/resources/cloud.properties
Validate that you see the correct ticket sale counts per title in the aggregation-sum-output topic.
confluent kafka topic consume aggregation-sum-output -b --print-key --delimiter :
You should see:
Guardians of the Galaxy:2 tickets sold
Doctor Strange:1 tickets sold
When you are finished, delete the kafka-streams-aggregating-sum-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>
git clone git@github.com:confluentinc/tutorials.git
cd tutorials
Start Kafka with the following command run from the top-level tutorials repository directory:
docker compose -f ./docker/docker-compose-kafka.yml up -d
Open a shell in the broker container:
docker exec -it broker /bin/bash
Create the input and output topics for the application:
kafka-topics --bootstrap-server localhost:9092 --create --topic aggregation-sum-input
kafka-topics --bootstrap-server localhost:9092 --create --topic aggregation-sum-output
Start a console producer:
kafka-console-producer --bootstrap-server localhost:9092 --topic aggregation-sum-input
Enter a few JSON-formatted ticket sales:
{"title":"Guardians of the Galaxy", "ticketTotalValue":15}
{"title":"Doctor Strange", "ticketTotalValue":15}
{"title":"Guardians of the Galaxy", "ticketTotalValue":15}
Enter Ctrl+C to exit the console producer.
On your local machine, compile the app:
./gradlew aggregating-sum:kstreams:shadowJar
Navigate into the application's home directory:
cd aggregating-sum/kstreams
Run the application, passing the local.properties Kafka client configuration file that points to the broker's bootstrap servers endpoint at localhost:9092:
java -cp ./build/libs/aggregating-sum-standalone.jar \
io.confluent.developer.KafkaStreamsAggregatingCount \
./src/main/resources/local.properties
Validate that you see the correct ticket sale counts per title in the aggregation-sum-output topic. In the broker container shell:
kafka-console-consumer --bootstrap-server localhost:9092 --topic aggregation-sum-output --from-beginning --property "print.key=true" --property "key.separator=:"
You should see:
Doctor Strange:1 tickets sold
Guardians of the Galaxy:2 tickets sold
From your local machine, stop the broker container:
docker compose -f ./docker/docker-compose-kafka.yml down