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

Tutorial

How to join a stream and a table in ksqlDB

ksqlDBKafka Streams
‹ Back to Tutorials

How to join a stream and a table in ksqlDB

Suppose that you have events in a Kafka topic and a table of reference data (also known as a lookup table). Let's see how you can join each event in the stream to attributes in the table based on a common key.

Setup

Let's use the example of a movie rating event stream. But the stream only contains the movie id, which isn't very descriptive, so you want to enrich it with some additional information. So you'll set up join between the stream and a table that contains fact or lookup data.

Here's the movie rating stream:

CREATE STREAM ratings (movie_id INT KEY, rating DOUBLE)
  WITH (KAFKA_TOPIC='ratings', 
        PARTITIONS=1, 
        VALUE_FORMAT='JSON');

And this is the table definition containing the movie reference data:

CREATE TABLE movies (id INT PRIMARY KEY, title VARCHAR, release_year INT)
    WITH (KAFKA_TOPIC='movies', 
          PARTITIONS=1, 
          VALUE_FORMAT='JSON');

Note that for a stream-table join to succeed, the primary key of the table must be the key of the stream. For example, the movies primary key id matches up with the ratings stream key of movie_id.

With your stream and table in place you can build a join like this:

CREATE STREAM rated_movies AS
    SELECT ratings.movie_id AS id, title, rating
    FROM ratings
    LEFT JOIN movies ON ratings.movie_id = movies.id;

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 ratings stream and movies table backed by Kafka running in Docker and populate them with test data.

CREATE STREAM ratings (movie_id INT KEY, rating DOUBLE)
  WITH (KAFKA_TOPIC='ratings', 
        PARTITIONS=1, 
        VALUE_FORMAT='JSON');
CREATE TABLE movies (id INT PRIMARY KEY, title VARCHAR, release_year INT)
    WITH (KAFKA_TOPIC='movies', 
          PARTITIONS=1, 
          VALUE_FORMAT='JSON');
INSERT INTO movies (id, title, release_year) VALUES (294, 'Twisters', 2024);
INSERT INTO movies (id, title, release_year) VALUES (354, 'Unfrosted', 2024);
INSERT INTO movies (id, title, release_year) VALUES (782, 'Family Switch', 2023);

INSERT INTO ratings (movie_id, rating) VALUES (294, 8.2);
INSERT INTO ratings (movie_id, rating) VALUES (294, 8.5);
INSERT INTO ratings (movie_id, rating) VALUES (354, 9.9);
INSERT INTO ratings (movie_id, rating) VALUES (354, 9.7);
INSERT INTO ratings (movie_id, rating) VALUES (782, 7.8);
INSERT INTO ratings (movie_id, rating) VALUES (782, 7.7);
INSERT INTO ratings (movie_id, rating) VALUES (782, 2.1);

Finally, run the stream-table join query and land the results in a new rated_movies stream. Note that we first tell ksqlDB to consume from the beginning of the streams.

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

CREATE STREAM rated_movies AS
    SELECT ratings.movie_id AS id, title, rating
    FROM ratings
    LEFT JOIN movies ON ratings.movie_id = movies.id
    EMIT CHANGES;

Query the new stream:

SELECT *
FROM rated_movies
EMIT CHANGES;

The query output should look like this:

+----------------------+----------------------+----------------------+
|ID                    |TITLE                 |RATING                |
+----------------------+----------------------+----------------------+
|294                   |Twisters              |8.2                   |
|294                   |Twisters              |8.5                   |
|354                   |Unfrosted             |9.9                   |
|354                   |Unfrosted             |9.7                   |
|782                   |Family Switch         |7.8                   |
|782                   |Family Switch         |7.7                   |
|782                   |Family Switch         |2.1                   |
+----------------------+----------------------+----------------------+

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.