Exercise - Working with lakehouses in a streaming world

In the previous exercise, we used Apache Flink® to perform real-time streaming transformations on our kart racing game data. While Flink excels at real-time processing, there are scenarios where data scientists and analysts need to access this streaming data through familiar interfaces like SQL queries against traditional tables.

In this exercise, we'll use Tableflow to expose our Kafka topics as Iceberg tables, creating a bridge between our streaming data and traditional analytics tools. This allows data scientists to query real-time data using familiar SQL interfaces while maintaining the performance benefits of modern data lakehouse architectures.

What is Tableflow?

Tableflow is a Confluent feature that automatically creates and maintains Iceberg tables from Kafka topics. It provides:

• Near real-time data synchronization from Kafka topics to Iceberg tables
• Schema evolution support that handles changes in the data structure
• Data scientist-friendly interfaces for accessing streaming data
• Performance optimization through columnar storage and partitioning

Producing data

Tableflow can only materialize data that is still in our topics, and we set the topics to keep data for only one hour. If the data generator isn't already running, start one of the scripts now so the topics contain fresh data:

./generate_data_local.sh

or

./generate_data_docker.sh

Leave it running for the rest of the exercise.

Setting up Tableflow

Now we'll set up Tableflow to start syncing our Kafka topics to Iceberg tables. Before we begin, ensure the following are in place:

• A Confluent Cloud account with the data_engineering environment active
• The exercises cluster running with our game data topics
• A compute pool configured for our environment

Enable Tableflow on the topics

We'll enable Tableflow on the three main topics from our game data: players, games, and player_activity.

confluent tableflow topic enable games
confluent tableflow topic enable players
confluent tableflow topic enable player_activity

Verify the Tableflow setup

Let's verify that Tableflow is running correctly. Check that the Tableflow instances are running:

confluent tableflow topic list

The output of this command is rather verbose. We can use something like awk to filter out columns we do not want:

confluent tableflow topic list | awk -F'[|+]' '{print $2,$15}'

This should show three Tableflow instances, one for each topic, with status "RUNNING".

Set up access to the Iceberg catalog on Confluent Cloud

Trino will need access to the catalog to get the list of tables and their properties. These properties were already written to tableflow.properties by the quickstart command we ran in the Confluent Cloud setup exercise. We'll need four (4) parameters to populate the Trino config file:

• Tableflow access key
• Tableflow secret
• Iceberg Catalog REST Endpoint URI
• Cloud region

If you prefer to enter these values manually, copy the iceberg.properties.template file to iceberg.properties in the trino-config/catalog/ folder and fill in the values. Alternatively, run the following script to do it automatically. It simply copies the needed properties to the Trino config file.

./update-config.sh

Query the Iceberg tables

Now that Tableflow is creating Iceberg tables from our Kafka topics, we can query them using familiar SQL interfaces. Let's explore some practical queries that demonstrate the power of this approach.

Bring up the environment

Bring up the Trino docker container:

docker compose up -d

Then run the trino CLI:

docker compose exec trino trino --server http://localhost:8080

Alternatively, run the following script to bring up Trino and put us in the CLI:

./run_trino_docker.sh

Setting the session schema in Trino

Tableflow creates a schema for each Kafka cluster with the Kafka cluster ID as the schema name. A Kafka cluster ID is usually prefixed with lkc-. Let's set it as our session schema to avoid having to prefix it to each table name:

SHOW CATALOGS; -- this will give iceberg
SHOW SCHEMAS FROM iceberg; -- this will give the Trino schema name
USE iceberg."<your kafka cluster id>";
SHOW TABLES;

Since the cluster ID has a -, the double quotes are necessary.

Basic data check

First, let's verify that our data is being synchronized correctly.

Note: To create Iceberg tables, Tableflow may take up to 15 minutes to write out the file. This is a choice to balance data freshness with cost efficiency. It may change in the future.

A simple check is in sql/tableflow_check_players.sql. It counts the number of players from the players table. Copy the SQL into the Trino CLI to run it.

Tableflow fields

In addition to the fields in the Kafka topic, Tableflow automatically adds Kafka metadata to each table in Iceberg. We can see them prefixed with $$ when we see the table definition:

DESCRIBE players;

This shows the topic, the partition, the offset and so on.

Querying the data lakehouse

Now let's explore real-world analytical queries that demonstrate the power of combining streaming data with lakehouse architecture. We'll run several queries that help answer important business questions.

Game popularity analysis

Track designers want to know which combination of track and game type is the most popular so they can make more of those. One way for us to do that is to check the average number of players in each game. Review the SQL in sql/tableflow_game_popularity_analysis.sql and then run it in the CLI.

Correlation between completion and off-track behavior

The Product Management team wants to increase player engagement. One idea is to make it easier to come back into a race after having gone off track. They plan to redesign the tracks and need a list of tracks and their average time to completion from an off-track event. The SQL in sql/tableflow_offtrack_recovery_times.sql analyzes how long it takes players to finish a race after going off track, helping identify which tracks have the most punishing recovery times. Review the query and then run it in the CLI.

Tracks that need to be better designed

Players usually drop out of games if they do not improve over time. The product management team wants a list of tracks that are badly designed. We can detect those if the player is generally improving on all tracks but not on that specific one. The SQL in sql/tableflow_improve_track_design.sql uses linear regression to compare each player's improvement on specific tracks versus their overall improvement trend, flagging tracks where players aren't getting better despite improving elsewhere. Review this more complex analytical query and then run it in the CLI. While the SQL code is rather large, it isn't earth-shattering:

• player_games creates a list of players, tracks, their win position and generates a sequence number for all the player's games
• player_track_performance - computes the player's specific improvement per track
• player_overall_performance - computes the player's improvement across all tracks
• track_quality - computes the average difference between overall improvement and that specific track's for all players
• Finally, we print out the report with some thresholds for consideration

While something like this could be done in Flink, it would involve careful state management. In a lakehouse, we can run this SQL without worrying and let the query engine figure things out. Note also the performance from using a subset of fields; something in which the columnar storage of Iceberg excels.

Finish

We've successfully set up Tableflow to create Iceberg tables from our Kafka topics and explored various ways to query this real-time data using familiar SQL interfaces. Tableflow provides a seamless bridge between Kafka topics and Iceberg tables, making near real-time data accessible to data scientists through familiar SQL interfaces. Performance is optimized through columnar storage and automatic partitioning, while integration with analytics tools remains straightforward and flexible.

This hybrid approach allows us to leverage the real-time capabilities of Kafka while providing the analytical power and accessibility of modern data lakehouse architectures. This brings us to the end of the exercise.

Resources

• Tableflow in Confluent Cloud
• Apache Iceberg Documentation
• Confluent CLI Reference
• Multi-Engine Support - Apache Iceberg
• Trino Iceberg Connector