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

Tutorial

How to filter messages in a Kafka topic with ksqlDB

Flink Table API (Java)Flink Table API (Python)Flink SQLksqlDBKafka Streams
‹ Back to Tutorials

How to filter messages in a Kafka topic with ksqlDB

How do you filter messages in a Kafka topic to contain only those that you're interested in? In this tutorial, we will filter a stream of book publications down to those by a particular author.

Setup

First, let's create a base stream of events containing book publications:

CREATE STREAM all_publications (bookid BIGINT KEY, 
                                author VARCHAR, 
                                title VARCHAR)
    WITH (KAFKA_TOPIC='publication_events'
          PARTITIONS=1,
          VALUE_FORMAT='JSON');

To create a new topic containing only events for a particular author, we issue a CREATE STREAM AS SELECT query containing the appropriate WHERE clause:

CREATE STREAM george_martin WITH (KAFKA_TOPIC='george_martin_books') AS
    SELECT *
      FROM all_publications
      WHERE author = 'George R. R. Martin';

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

CREATE STREAM all_publications (book_id BIGINT KEY, 
                                author VARCHAR, 
                                title VARCHAR)
    WITH (KAFKA_TOPIC='publication_events',
          PARTITIONS=1,
          VALUE_FORMAT='JSON');
INSERT INTO all_publications (book_id, author, title) VALUES (1, 'C.S. Lewis', 'The Silver Chair');
INSERT INTO all_publications (book_id, author, title) VALUES (2, 'George R. R. Martin', 'A Song of Ice and Fire');
INSERT INTO all_publications (book_id, author, title) VALUES (3, 'C.S. Lewis', 'Perelandra');
INSERT INTO all_publications (book_id, author, title) VALUES (4, 'George R. R. Martin', 'Fire & Blood');
INSERT INTO all_publications (book_id, author, title) VALUES (5, 'J. R. R. Tolkien', 'The Hobbit');
INSERT INTO all_publications (book_id, author, title) VALUES (6, 'J. R. R. Tolkien', 'The Lord of the Rings');
INSERT INTO all_publications (book_id, author, title) VALUES (7, 'George R. R. Martin', 'A Dream of Spring');
INSERT INTO all_publications (book_id, author, title) VALUES (8, 'J. R. R. Tolkien', 'The Fellowship of the Ring');
INSERT INTO all_publications (book_id, author, title) VALUES (9, 'George R. R. Martin', 'The Ice Dragon');

Finally, run the filter query to find books by George R. R. Martin and write the events to a new topic. Note that we first tell ksqlDB to consume from the beginning of the stream.

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

CREATE STREAM george_martin WITH (KAFKA_TOPIC='george_martin_books') AS
  SELECT *
    FROM all_publications
    WHERE author = 'George R. R. Martin';

Query the new topic:

SELECT *
FROM george_martin
EMIT CHANGES;

The query output should look like this:

+-------------------------------+-------------------------------+-------------------------------+
|BOOK_ID                        |AUTHOR                         |TITLE                          |
+-------------------------------+-------------------------------+-------------------------------+
|2                              |George R. R. Martin            |A Song of Ice and Fire         |
|4                              |George R. R. Martin            |Fire & Blood                   |
|7                              |George R. R. Martin            |A Dream of Spring              |
|9                              |George R. R. Martin            |The Ice Dragon                 |
+-------------------------------+-------------------------------+-------------------------------+

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.