How to count the number of messages in a topic with ksqlDB

It can be useful to know how many messages are currently in a topic, but you cannot calculate this directly based on the offsets, because you need to consider the topic's retention policy, log compaction, and potential duplicate messages. In this example, we'll take a topic of pageview data and see how we can count all the messages in the topic. Note that the time complexity for this tutorial is O(n) (linear); processing time will depend on the number of messages in the topic, and large data sets will require long running times.

Setup

First, create a stream over the topic you're interested in counting the number of records.

CREATE STREAM pageviews (msg VARCHAR)
  WITH (KAFKA_TOPIC ='pageviews',
        VALUE_FORMAT='JSON');

Note that at this stage we’re just interested in counting the messages in their entirety, so we define the loosest schema possible, msg VARCHAR, for speed. Also, you'll need to configure ksqlDB to start from the beginning of the topic so that all messages are included in the count:

SET 'auto.offset.reset' = 'earliest';

Then COUNT the events in the pageviews stream:

SELECT COUNT(*) AS msg_count
FROM pageviews
EMIT CHANGES;

The query above will run continually until you cancel it.

Running the example

Prerequisites

• Docker running via Docker Desktop or Docker Engine
• Docker Compose. Ensure that the command docker compose version succeeds.

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 pageviews stream backed by Kafka running in Docker and populate it with test data.

CREATE STREAM pageviews (msg VARCHAR)
  WITH (KAFKA_TOPIC ='pageviews',
        PARTITIONS=1,
        VALUE_FORMAT='JSON');

INSERT INTO pageviews (msg) VALUES ('https://www.acme.com/home');
INSERT INTO pageviews (msg) VALUES ('https://www.acme.com/search');
INSERT INTO pageviews (msg) VALUES ('https://www.acme.com/home');
INSERT INTO pageviews (msg) VALUES ('https://www.acme.com/products');

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 count result.

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

SELECT COUNT(*) AS msg_count
FROM pageviews
EMIT CHANGES;

The query output should look like this:

+----------------------------------------+
|MSG_COUNT                               |
+----------------------------------------+
|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