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

Tutorial

How to count the number of events in a Kafka topic with ksqlDB

Flink SQLKafka StreamsksqlDB
‹ Back to Tutorials

How to count the number of events in a Kafka topic with ksqlDB

Suppose you have a topic with events that represent ticket sales for movies. In this tutorial, we will use ksqlDB to calculate the total number of tickets sold per movie.

Setup

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

CREATE STREAM movie_ticket_sales (title VARCHAR, sale_ts VARCHAR, ticket_total_value INT)
    WITH (KAFKA_TOPIC='movie-ticket-sales',
          PARTITIONS=1,
          VALUE_FORMAT='AVRO');

Compute the count

Given the movie_ticket_sales stream definition above, we can figure out the total number of tickets sold per movie using the following COUNT aggregation:

SELECT title,
       COUNT(*) AS tickets_sold
FROM movie_ticket_sales
GROUP BY title
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_ticket_sales stream backed by Kafka running in Docker and populate it with test data.

CREATE STREAM movie_ticket_sales (title VARCHAR, sale_ts VARCHAR, ticket_total_value INT)
    WITH (KAFKA_TOPIC='movie-ticket-sales',
          PARTITIONS=1,
          VALUE_FORMAT='AVRO');
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Unfrosted', '2024-09-18T10:00:00Z', 10);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Family Switch', '2024-09-18T10:00:00Z', 12);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Family Switch', '2024-09-18T10:01:00Z', 12);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Twisters', '2024-09-18T10:01:31Z', 12);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Family Switch', '2024-09-18T10:01:36Z', 24);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Twisters', '2024-09-18T10:02:00Z', 18);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Twisters', '2024-09-18T11:40:00Z', 36);
INSERT INTO movie_ticket_sales (title, sale_ts, ticket_total_value) VALUES ('Twisters', '2024-09-18T11:40:09Z', 18);

Finally, run the aggregating count 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 (movie title).

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

SELECT title,
       COUNT(*) AS tickets_sold
FROM movie_ticket_sales
GROUP BY title
EMIT CHANGES;

The query output should look like this:

+----------------------------------+----------------------------------+
|TITLE                             |TICKETS_SOLD                      |
+----------------------------------+----------------------------------+
|Unfrosted                         |1                                 |
|Family Switch                     |3                                 |
|Twisters                          |4                                 |
+----------------------------------+----------------------------------+

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.