How to transform events with Kafka Streams

If you have a stream of events in a Kafka topic and wish to transform a field in each event, you simply need to use the KStream.map method to process each event.

As a concrete example, consider a topic with events that represent movies. Each event has a single attribute that combines its title and its release year into a string. The following topology definition outputs these events to a new topic with title and release date turned into their own attributes.

  builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), rawMovieSerde))
    .map((key, rawMovie) -> new KeyValue<>(rawMovie.id(), convertRawMovie(rawMovie)))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.Long(), movieSerde));

The map method transforms each record of the input stream into a new record in the output stream, with the movie ID serving as the key. (There is also a mapValues method that can be used if you only need to transform record values.) The convertRawMovie method in this example splits on :: since the attribute containing the movie title and release year looks like Tree of Life::2011. The Serde's included in this example use Jackson to serialize and deserialize POJOs.

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-transforming-env \
  --kafka-cluster-name kafka-streams-transforming-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./transforming/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 raw-movies
confluent kafka topic create movies

Start a console producer:

confluent kafka topic produce raw-movies

Enter a few JSON-formatted movies in which the title field has the release year appended:

{"id":"294", "title":"Die Hard::1988", "genre":"action"}
{"id":"354", "title":"Tree of Life::2011", "genre":"drama"}
{"id":"782", "title":"A Walk in the Clouds::1995", "genre":"romance"}
{"id":"128", "title":"The Big Lebowski::1998", "genre":"comedy"}

Enter Ctrl+C to exit the console producer.

Compile and run the application

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

./gradlew transforming:kstreams:shadowJar

Navigate into the application's home directory:

cd transforming/kstreams

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

java -cp ./build/libs/transforming-standalone.jar \
    io.confluent.developer.TransformStream \
    ./src/main/resources/cloud.properties

Validate that you see movies with proper title and release_year fields in the movies topic.

confluent kafka topic consume movies -b

You should see:

{"id":294,"title":"Die Hard","release_year":1988,"genre":"action"}
{"id":354,"title":"Tree of Life","release_year":2011,"genre":"drama"}
{"id":782,"title":"A Walk in the Clouds","release_year":1995,"genre":"romance"}
{"id":128,"title":"The Big Lebowski","release_year":1998,"genre":"comedy"}

Clean up

When you are finished, delete the kafka-streams-transforming-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 compose -f ./docker/docker-compose-kafka.yml up -d

docker exec -it broker /bin/bash

kafka-topics --bootstrap-server localhost:9092 --create --topic raw-movies
kafka-topics --bootstrap-server localhost:9092 --create --topic movies

kafka-console-producer --bootstrap-server localhost:9092 --topic raw-movies

{"id":"294", "title":"Die Hard::1988", "genre":"action"}
{"id":"354", "title":"Tree of Life::2011", "genre":"drama"}
{"id":"782", "title":"A Walk in the Clouds::1995", "genre":"romance"}
{"id":"128", "title":"The Big Lebowski::1998", "genre":"comedy"}

./gradlew transforming:kstreams:shadowJar

cd transforming/kstreams

java -cp ./build/libs/transforming-standalone.jar \
    io.confluent.developer.TransformStream \
    ./src/main/resources/local.properties

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

{"id":294,"title":"Die Hard","release_year":1988,"genre":"action"}
{"id":354,"title":"Tree of Life","release_year":2011,"genre":"drama"}
{"id":782,"title":"A Walk in the Clouds","release_year":1995,"genre":"romance"}
{"id":128,"title":"The Big Lebowski","release_year":1998,"genre":"comedy"}

docker compose -f ./docker/docker-compose-kafka.yml down