Get Started Free
Tutorial

How to join a KStream and a KTable in Kafka Streams

ksqlDBKafka Streams
‹ Back to Tutorials

How to join a KStream and a KTable in Kafka Streams

Suppose you have a set of movies that have been released and a stream of ratings from moviegoers about how entertaining they are. In this tutorial, we'll write a program that joins each rating with content about the movie.

First you'll create a KTable for the reference movie data

KTable<Long, Movie> movieTable = builder.stream(MOVIE_INPUT_TOPIC,
                Consumed.with(Serdes.String(), movieSerde))
        .peek((key, value) -> LOG.info("Incoming movies key[{}] value[{}]", key, value))
        .toTable(Materialized.with(Serdes.String(), movieSerde));

Here you've started with a KStream so you can add a peek statement to view the incoming records, then convert it to a table with the toTable operator. Otherwise, you could create the table directly with StreamBuilder.table. Here we are assuming the underlying topic is keyed with the movie id.

Then you'll create your KStream of ratings:

 KStream<Long, Rating> ratings = builder.stream(RATING_INPUT_TOPIC,
                        Consumed.with(Serdes.Long(), ratingSerde))
                .map((key, rating) -> new KeyValue<>(rating.id(), rating));

We need to have the same id as the table, so we'll use a KStream.map operator to set the rating id as the key. The Rating class id uses the same id as the movie.

Now you use a ValueJoiner specifying how to construct the joined value of the stream and table:

public class MovieRatingJoiner implements ValueJoiner<Rating, Movie, RatedMovie> {

  public RatedMovie apply(Rating rating, Movie movie) {
    return new RatedMovie(movie.id(), movie.title(), movie.releaseYear(), rating.rating());
  }
}

Now, you'll put all this together using a KStream.join operation:

 ratings.join(movieTable, joiner, Joined.with(Serdes.Long(),ratingSerde, movieSerde))
                .to(RATED_MOVIES_OUTPUT, Produced.with(Serdes.Long(), ratedMovieSerde));

Notice that you're supplying the Serde for the key, the stream value and the value of the table via the Joined configuration object.

The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.

Prerequisites

  • A Confluent Cloud account
  • The Confluent CLI installed on your machine
  • Apache Kafka or Confluent Platform (both include the Kafka Streams application reset tool)
  • Clone the confluentinc/tutorials repository and navigate into its top-level directory: 
    git clone git@github.com:confluentinc/tutorials.git
cd tutorials

Create Confluent Cloud resources

Login to your Confluent Cloud account:

confluent login --prompt --save

Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:

confluent plugin install confluent-quickstart

Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial. Note that you may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.

confluent quickstart \
  --environment-name kafka-streams-stream-table-join-env \
  --kafka-cluster-name kafka-streams-stream-table-join-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./joining-stream-table/kstreams/src/main/resources/cloud.properties

The plugin should complete in under a minute.

Create topics

Create the input and output topics for the application:

confluent kafka topic create movie-input
confluent kafka topic create ratings-input
confluent kafka topic create rated-movies-output

Start a console producer:

confluent kafka topic produce movie-input --parse-key --delimiter :

Enter a few JSON-formatted movie objects:

294:{"id":"294", "title":"Die Hard", "releaseYear":1988}
354:{"id":"354", "title":"Tree of Life", "releaseYear":2011}
782:{"id":"782", "title":"A Walk in the Clouds", "releaseYear":1998}
128:{"id":"128", "title":"The Big Lebowski", "releaseYear":1998}
780:{"id":"780", "title":"Super Mario Bros.", "releaseYear":1993}

Enter Ctrl+C to exit the console producer.

Similarly, start a console producer for movie ratings:

confluent kafka topic produce ratings-input --parse-key --delimiter :

Enter a few JSON-formatted ratings:

294:{"id":"294", "rating":8.2}
294:{"id":"294", "rating":8.5}
354:{"id":"354", "rating":9.9}
354:{"id":"354", "rating":9.7}
782:{"id":"782", "rating":7.8}
782:{"id":"782", "rating":7.7}
128:{"id":"128", "rating":8.7}
128:{"id":"128", "rating":8.4}
780:{"id":"780", "rating":2.1}

Enter Ctrl+C to exit the console producer.

Compile and run the application

Compile the application from the top-level tutorials repository directory:

./gradlew joining-stream-table:kstreams:shadowJar

Navigate into the application's home directory:

cd joining-stream-table/kstreams

Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:

java -cp ./build/libs/kstreams-stream-table-join-standalone.jar \
    io.confluent.developer.JoinStreamToTable \
    ./src/main/resources/cloud.properties

Validate that you see enriched movie ratings in the rated-movies-output topic.

confluent kafka topic consume rated-movies-output -b

You should see:

{"id":"294","title":"Die Hard","releaseYear":1988,"rating":8.2}
{"id":"294","title":"Die Hard","releaseYear":1988,"rating":8.5}
{"id":"354","title":"Tree of Life","releaseYear":2011,"rating":9.9}
{"id":"354","title":"Tree of Life","releaseYear":2011,"rating":9.7}
{"id":"782","title":"A Walk in the Clouds","releaseYear":1998,"rating":7.8}
{"id":"782","title":"A Walk in the Clouds","releaseYear":1998,"rating":7.7}
{"id":"128","title":"The Big Lebowski","releaseYear":1998,"rating":8.7}
{"id":"128","title":"The Big Lebowski","releaseYear":1998,"rating":8.4}
{"id":"780","title":"Super Mario Bros.","releaseYear":1993,"rating":2.1}

Clean up

When you are finished, delete the kafka-streams-stream-table-join-env environment by first getting the environment ID of the form env-123456 corresponding to it:

confluent environment list

Delete the environment, including all resources created for this tutorial:

confluent environment delete <ENVIRONMENT ID>
Docker instructions

Prerequisites

  • Docker running via Docker Desktop or Docker Engine
  • Docker Compose. Ensure that the command docker compose version succeeds.
  • Clone the confluentinc/tutorials repository and navigate into its top-level directory: 
    git clone git@github.com:confluentinc/tutorials.git
cd tutorials

Start Kafka in Docker

Start Kafka with the following command run from the top-level tutorials repository directory:

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

Create topics

Open a shell in the broker container:

docker exec -it broker /bin/bash

Create the input and output topics for the application:

kafka-topics --bootstrap-server localhost:9092 --create --topic movie-input
kafka-topics --bootstrap-server localhost:9092 --create --topic ratings-input
kafka-topics --bootstrap-server localhost:9092 --create --topic rated-movies-output

Start a console producer:

kafka-console-producer --bootstrap-server localhost:9092 --topic movie-input \
    --property "parse.key=true" --property "key.separator=:"

Enter a few JSON-formatted movie objects:

294:{"id":"294", "title":"Die Hard", "releaseYear":1988}
354:{"id":"354", "title":"Tree of Life", "releaseYear":2011}
782:{"id":"782", "title":"A Walk in the Clouds", "releaseYear":1998}
128:{"id":"128", "title":"The Big Lebowski", "releaseYear":1998}
780:{"id":"780", "title":"Super Mario Bros.", "releaseYear":1993}

Enter Ctrl+C to exit the console producer.

Similarly, start a console producer for movie ratings:

kafka-console-producer --bootstrap-server localhost:9092 --topic ratings-input \
    --property "parse.key=true" --property "key.separator=:"

Enter a few JSON-formatted ratings:

294:{"id":"294", "rating":8.2}
294:{"id":"294", "rating":8.5}
354:{"id":"354", "rating":9.9}
354:{"id":"354", "rating":9.7}
782:{"id":"782", "rating":7.8}
782:{"id":"782", "rating":7.7}
128:{"id":"128", "rating":8.7}
128:{"id":"128", "rating":8.4}
780:{"id":"780", "rating":2.1}

Enter Ctrl+C to exit the console producer.

Compile and run the application

On your local machine, compile the app:

./gradlew joining-stream-table:kstreams:shadowJar

Navigate into the application's home directory:

cd joining-stream-table/kstreams

Run the application, passing the local.properties Kafka client configuration file that points to the broker's bootstrap servers endpoint at localhost:9092:

java -cp ./build/libs/kstreams-stream-table-join-standalone.jar \
    io.confluent.developer.JoinStreamToTable \
    ./src/main/resources/local.properties

Validate that you see enriched movie ratings in the rated-movies-output topic. In the broker container shell:

kafka-console-consumer --bootstrap-server localhost:9092 --topic rated-movies-output --from-beginning

You should see:

{"id":"294","title":"Die Hard","releaseYear":1988,"rating":8.2}
{"id":"294","title":"Die Hard","releaseYear":1988,"rating":8.5}
{"id":"354","title":"Tree of Life","releaseYear":2011,"rating":9.9}
{"id":"354","title":"Tree of Life","releaseYear":2011,"rating":9.7}
{"id":"782","title":"A Walk in the Clouds","releaseYear":1998,"rating":7.8}
{"id":"782","title":"A Walk in the Clouds","releaseYear":1998,"rating":7.7}
{"id":"128","title":"The Big Lebowski","releaseYear":1998,"rating":8.7}
{"id":"128","title":"The Big Lebowski","releaseYear":1998,"rating":8.4}
{"id":"780","title":"Super Mario Bros.","releaseYear":1993,"rating":2.1}

Clean up

From your local machine, stop the broker container:

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