Exercise - Transforming data with Apache Flink®

In the previous exercise, we created a set of Kafka topics named players, games, and player_activity populated with data from our online kart racing game.

Next, we will transform that data, performing aggregations and enriching it to make it more useful for downstream consumers. We will do this using Flink SQL queries executed in Confluent Cloud.

Create a workspace

Before we can execute Flink SQL queries, we need to set up a workspace. Confluent Cloud uses workspaces to organize and execute Flink SQL queries.

Create a new workspace in Confluent Cloud using the compute pool we created for the data_engineering environment.

In the workspace, select the data_engineering catalog and the exercises database.

Note: Set the cloud and region to the same values we used when we set up our environment.

Command line reference

If you prefer to interact with Flink through the interactive shell provided by the Confluent Command Line Interface (CLI), use the following commands.

Find the ID of the compute pool:

confluent flink compute-pool list

Set the active compute pool:

confluent flink compute-pool use <ID>

Start an interactive Flink shell:

confluent flink shell

Producing data

We need data flowing into our topics before we can transform it. If the data generator isn't already running, we'll start it now.

If it's not already running, start one of the scripts to produce data to our source topics.

./generate_data_local.sh

or

./generate_data_docker.sh

Player completion statistics

Each player in the game has a profile. It contains details about their performance in past races, including:

• Best Finish
• Worst Finish
• Average Finish
• Total First Place Finishes

By default, the profile is only visible to the player, however, they can optionally share it with friends, or they can make it public. This allows players to show off their achievements.

To populate the profile, we need to aggregate data that has been captured in the player_activity table.

Players expect that when they complete a race, the results will be posted to their profile immediately. Therefore, it is critical that we perform these computations using a real-time data stream.

We have created a series of Flink SQL statements to help populate the data. We can find these queries in the GitHub repository we cloned in the previous exercise.

The statements are in the sql/player_completion_statistics.sql file.

They demonstrate how we can use Flink SQL to perform aggregations on streaming data in real-time.

Create the destination topic

Open the player_completion_statistics.sql file and locate the statement that begins with:

CREATE TABLE data_engineering.exercises.player_completion_statistics 

Review the statement, then copy it into the workspace and execute it to create a destination table/topic for the aggregated statistics.

Compute the player statistics

Next, locate the statement that starts with:

INSERT INTO data_engineering.exercises.player_completion_statistics

This statement will compute the aggregated statistics for the player.

Review the statement before copying it and executing it in the workspace.

Leave it running while moving on to the next part.

Observe the results

Finally, observe the results and verify the statement is working by executing:

SELECT * FROM data_engineering.exercises.player_completion_statistics;

This should show a series of aggregated results for each player.

Player rankings

With player statistics computed, we can now build on that data to create player rankings that help match players of similar skill levels.

Players will be matched based on their skill levels. This ensures that high-ranked players are not being matched with low-rank players.

In a real system, this ranking would be complex, potentially even being fed through an AI to determine the ranking.

In this course, we'll use a simpler approach. We will look at the top_3_finish_rate computed in the player_completion_statistics and apply a formula to assign players to one of three categories:

• beginner: top_3_finish_rate < 34
• intermediate: top_3_finish_rate between 34 and 66 (inclusive)
• expert: top_3_finish_rate > 66

We can find the statements for this in the sql/player_rankings.sql file.

They demonstrate how we can re-use data from other statements in order to create more complex streaming pipelines.

Create the destination topic

Start by creating the destination table. Find and execute the statement that begins with:

CREATE TABLE data_engineering.exercises.player_rankings

Compute the player rankings

Next, let's perform the computation. Find and execute the statement that begins with:

INSERT INTO data_engineering.exercises.player_rankings

Observe the results

Finally, observe the results by executing:

SELECT * FROM data_engineering.exercises.player_rankings;

Popular games

Now we'll help players find games to join by identifying the most popular games. After ranking, players are presented with games matching their skill level. They can then select from that list based on various criteria, such as games with the fewest players, or games with the most players, or games that appear to be more popular.

The most popular games query is the one we are going to implement next.

The records in the games table get pushed out whenever an update to the game occurs. To determine the most popular games, we will be grouping those records into one minute windows, based on the timestamp of the record. We'll then be looking for games where there are 3 or fewer open slots remaining.

Essentially, we're looking for the recently updated games that are mostly full.

We can find the statements for this in the sql/games_almost_full.sql file.

They demonstrate how we can use windows in Flink SQL to group records into windowed time periods based on their timestamp.

Create the destination topic

Start by creating the destination table. Find and execute the statement that begins with:

CREATE TABLE `data_engineering`.`exercises`.`games_almost_full`

Determine the popular games

Next, let's find the popular games. Find and execute the statement that begins with:

INSERT INTO `data_engineering`.`exercises`.`games_almost_full`

Observe the results

Finally, observe the results by executing:

SELECT * FROM data_engineering.exercises.games_almost_full;

Catching cheaters

Our final transformation will help identify potential cheaters by analyzing suspicious player behavior patterns. The game is designed to be family-friendly, but unfortunately, online games attract those who prefer to cheat. Recently, the game has experienced a small number of players who are leveraging various cheats that allow them to win games without developing the necessary skill.

There are various techniques that have been developed to isolate the cheaters so they can be dealt with appropriately. One technique monitors how often players fall off the track, but still manage to come in the top 3 for the race.

The idea behind this is that a player who is constantly falling off the track, but still manages to win, might be using some form of cheat. For example, they may have found a flaw in the track geometry that allows them to fall through the track and bypass portions of the map. Or they may have triggered some exploit code that allows them to fly.

Of course, it's also possible that the track is just poorly designed, and players are falling off way more than they should.

In any case, we want to flag these incidents for further analysis. This analysis might be performed by humans, but more likely we'll leverage some AI to compare various trends to find the cheaters in real-time.

We'll be looking for players who fall off the track 4 or more times in a one-minute window, but still manage to finish in the top 3.

We can find the statements for this in the sql/player_off_track_detection.sql file.

This is a reasonably complex operation. It demonstrates how Common Table Expressions (CTEs) created using the WITH keyword can be used to create intermediate tables that can be composed into the final query. In this case, the tables are being combined using a JOIN. Specifically, it uses an "Interval" join, which helps keep the query state from growing out of control. It also shows how windows and aggregations that we've seen previously can be combined in a single query.

Create the destination topic

Start by creating the destination table. Find and execute the statement that begins with:

CREATE TABLE data_engineering.exercises.player_off_track_detection

Determine the potential cheaters

Next, let's find the potential cheaters. Find and execute the statement that begins with:

INSERT INTO data_engineering.exercises.player_off_track_detection

Observe the results

Finally, observe the results by executing:

SELECT * FROM data_engineering.exercises.player_off_track_detection;

Finish

We've implemented four different queries showing various features in Flink SQL. They are operating in real-time or using short time windows so we can see quick results. However, windows can be longer if necessary, allowing Flink to compute in real-time the kinds of queries that would typically be executed as batches. Generating a monthly report in Flink can be as simple as setting the window time to one month instead of one minute.

This brings us to the end of the exercise. Feel free to stop the data generator. We can restart it when we reach the next exercise.