Enhance your career, get your certificate as a Data Streaming Engineer | Get your Certificate

Tutorial

How to compute the minimum or maximum value of a field with ksqlDB

Flink SQLksqlDBKafka Streams
‹ Back to Tutorials

How to compute the minimum or maximum value of a field with ksqlDB

Suppose you have a topic with events that represent ticket sales for movies. In this tutorial, we will use ksqlDB to compute the highest and lowest film revenues per year.

Setup

Let's assume the following DDL for our base movie_sales stream:

CREATE STREAM movie_sales (title VARCHAR, release_year INT, total_sales BIGINT)
    WITH (KAFKA_TOPIC='movie-sales',
          PARTITIONS=1,
          VALUE_FORMAT='AVRO');

Computing the min / max

Given the movie_sales stream definition above, we can figure out the total number of tickets sold per movie using the following minimum and maximum aggregation:

SELECT release_year,
       MIN(total_sales) AS min_total_sales,
       MAX(total_sales) AS max_total_sales
FROM movie_sales
GROUP BY release_year
    EMIT CHANGES;

Running the example

Prerequisites

Run the commands

Clone the confluentinc/tutorials GitHub repository (if you haven't already) and navigate to the tutorials directory:

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

Start ksqlDB and Kafka:

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

Next, open the ksqlDB CLI:

docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

Run the following SQL statements to create the movie_sales stream backed by Kafka running in Docker and populate it with test data.

CREATE STREAM movie_sales (title VARCHAR, release_year INT, total_sales BIGINT)
    WITH (KAFKA_TOPIC='movie-sales',
          PARTITIONS=1,
          VALUE_FORMAT='AVRO');
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Twisters', 2024, 369712130);
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Deadpool & Wolverine', 2024, 1318259708);
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Oppenheimer', 2023, 975579184);
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Barbie', 2023, 1445638421);
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Avatar: The Way of Water', 2022, 2320250281);
INSERT INTO movie_sales (title, release_year, total_sales) VALUES ('Jurassic World Dominion', 2022, 1001978080);

Finally, run the aggregating min / max query. Note that we first tell ksqlDB to consume from the beginning of the stream, and we also configure the query to use caching so that we only get a single output record per key ().

SET 'auto.offset.reset'='earliest';
SET 'ksql.streams.cache.max.bytes.buffering' = '10000000';

SELECT release_year, 
       MIN(total_sales) AS min_total_sales, 
       MAX(total_sales) AS max_total_sales 
FROM movie_sales
GROUP BY release_year
EMIT CHANGES;

The query output should look like this:

+--------------------------+--------------------------+--------------------------+
|RELEASE_YEAR              |MIN_TOTAL_SALES           |MAX_TOTAL_SALES           |
+--------------------------+--------------------------+--------------------------+
|2024                      |369712130                 |1318259708                |
|2023                      |975579184                 |1445638421                |
|2022                      |1001978080                |2320250281                |
+--------------------------+--------------------------+--------------------------+

When you are finished, exit the ksqlDB CLI by entering CTRL-D and clean up the containers used for this tutorial by running:

docker compose -f ./docker/docker-compose-ksqldb.yml down
Do you have questions or comments? Join us in the #developer-confluent-io community Slack channel to engage in discussions with the creators of this content.