All required resources in Confluent Cloud must be already created for this lab to work correctly. If you haven't already, please follow the prerequisites.
1. Verify Confluent Cloud Resources
Let's verify if all resources were created correctly and we can start using them.
Check if the following topics exist in your Kafka cluster:
- shoe_products (for product data aka Product Catalog),
- shoe_customers (for customer data aka Customer CRM),
- shoe_orders (for realtime order transactions aka Billing System).
Check if the following Avro schemas exist in your Schema Registry:
- shoe_products-value,
- shoe_customers-value,
- shoe_orders-value.
NOTE: Schema Registry is at the Environment level and can be used for multiple Kafka clusters.
Your Kafka cluster should have three Datagen Source Connectors running. Check if their topic and template configurations match the table below.
| Connector Name (can be anything) | Topic | Format | Template |
|---|---|---|---|
| DatagenSourceConnector_products | shoe_products | AVRO | Shoes |
| DatagenSourceConnector_customers | shoe_customers | AVRO | Shoes customers |
| DatagenSourceConnector_orders | shoe_orders | AVRO | Shoes orders |
IMPORTANT TO KNOW FOR THE WORKSHOP: We run in AWS only. Currently we do support 4 Regions within AWS cloud.
The complete onsite team is working in region: eu-central-1. (No changes in the guides (both Terraform and manual) necessary.)
The online team is working in different regions:
- Attendees with Lastname first Letter A-I working in region
us-east1- Flink SQL Pool in
us-east1
- Flink SQL Pool in
- Attendees with Lastname first Letter J-R working in region
us-east2- Flink SQL Pool in
us-east2
- Flink SQL Pool in
- Attendees with Lastname first Letter S-Z working in region
eu-west-1- Flink SQL Pool in
eu-west-1
- Flink SQL Pool in
Create a Flink Compute Pool in environment handson-flink. Now go back to environment handson-flink and choose the Flink (preview) Tab. From there we create a new compute pool:
- choose AWS region (remember the Lastname rule above), click
continueand - enter Pool Name:
cc_flink_compute_poolwith 5 Confluent Flink Units (CFU) and - click
Continuebutton and thenFinish. The pool will be provisioned and ready to use in a couple of moments.
You can use your web browser or console to enter Flink SQL statements.
-
Web UI - click on the button
Open SQL workspaceon your Flink Compute Pool You now have to set which catalog and database you want to use:- Set the environment name
handson-flinkas catalog, - and the cluster name
cc_handson_clusteras database
via the dropdown boxes, see graphic below
- Set the environment name
-
Console - copy/paste the command from your Flink Compute Pool to the command line.
Of course you could also use the the Flink SQL Shell. For this, you need to have Confluent Cloud Console tool installed and be logged in with correct access rights. Copy the command out of the Compute Pool Window and execute it in your terminal (we prefer iterm2).
confluent flink shell --compute-pool <pool id> --environment <env-id>NOTE: You can also access your Flink Compute Pool from the Data Portal as shown below. Just click on Data Portal in the main menu on the left side. Then select your Environment. You should see your topics. When you click on any of the topic tiles you can query the topic's data using Flink.
Data Portal: Kafka Topics Tiles
Data Portal: shoe_order topic selected. Click on Query button to access your Flink Compute Pool.
Let's start with exploring our Flink tables. Kafka topics and schemas are always in sync with our Flink cluster. Any topic created in Kafka is visible directly as a table in Flink, and any table created in Flink is visible as a topic in Kafka. Effectively, Flink provides a SQL interface on top of Confluent Cloud.
Following mappings exist:
| Kafka | Flink |
|---|---|
| Environment | Catalog |
| Cluster | Database |
| Topic + Schema | Table |
We will now work with the following SQL Worksheet:
Make sure you work with the correct Flink catalog (=environment) and database (=Kafka cluster).
SHOW CATALOGS;
SHOW DATABASES;
USE CATALOG <MY CONFLUENT ENVIRONMENT NAME>;
USE cc_handson_cluster;
List all Flink Tables (=Kafka topics) in your Confluent Cloud cluster:
SHOW TABLES;
Do you see tables shoe_products, shoe_customers, shoe_orders?
You can add multiple query boxes by clicking the + button on the left.
Understand how the table shoe_products was created:
SHOW CREATE TABLE shoe_products;
You can find more information about all parameters here.
Our Flink tables are populated by the Datagen connectors.
Let us first check the table schema for our shoe_products catalog. This should be the same as the topic schema in Schema Registry.
DESCRIBE shoe_products;
Let's check if any product records exist in the table.
SELECT * FROM shoe_products;
Now check if the shoe_customers schema exists.
DESCRIBE shoe_customers;
Are there any customers in Texas whose name starts with B ?
SELECT * FROM shoe_customers
WHERE `state` = 'Texas' AND `last_name` LIKE 'B%';
Check all attributes of the shoe_orders table including hidden attributes.
DESCRIBE EXTENDED shoe_orders;
Check the first ten orders for one customer.
SELECT order_id, product_id, customer_id, $rowtime
FROM shoe_orders
WHERE customer_id = 'b523f7f3-0338-4f1f-a951-a387beeb8b6a'
LIMIT 10;
Let's try to run more advanced queries.
First find out the number of customers records and then the number of unique customers.
SELECT COUNT(id) AS num_customers FROM shoe_customers;
SELECT COUNT(DISTINCT id) AS num_customers FROM shoe_customers;
We can try some basic aggregations with the product catalog records. For each shoe brand, find the number of shoe models, average rating and maximum model price.
SELECT brand as brand_name,
COUNT(DISTINCT name) as models_by_brand,
ROUND(AVG(rating),2) as avg_rating,
MAX(sale_price)/100 as max_price
FROM shoe_products
GROUP BY brand;
Let's try Flink's time windowing functions for shoe order records. Column names “window_start” and “window_end” are commonly used in Flink's window operations, especially when dealing with event time windows.
Find the amount of orders for one minute intervals (tumbling window aggregation).
SELECT
window_end,
COUNT(DISTINCT order_id) AS num_orders
FROM TABLE(
TUMBLE(TABLE shoe_orders, DESCRIPTOR(`$rowtime`), INTERVAL '1' MINUTES))
GROUP BY window_end;
Find the amount of orders for ten minute intervals advanced by five minutes (hopping window aggregation).
SELECT
window_start, window_end,
COUNT(DISTINCT order_id) AS num_orders
FROM TABLE(
HOP(TABLE shoe_orders, DESCRIPTOR(`$rowtime`), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
GROUP BY window_start, window_end;
NOTE: You can find more information about Flink Window aggregations here.
Flink allows you to define a primary key for your table. The primary key is a column whose value is unique for each record.
Let's create a new table that will store unique customers only.
CREATE TABLE shoe_customers_keyed(
customer_id STRING,
first_name STRING,
last_name STRING,
email STRING,
PRIMARY KEY (customer_id) NOT ENFORCED
);
Compare the new table shoe_customers_keyed with shoe_customers, what is the difference?
SHOW CREATE TABLE shoe_customers_keyed;We do have a different changelog.mode and a primary key constraint. What does this mean?
NOTE: You can find more information about primary key constraints here.
Create a new Flink job to copy customer data from the original table to the new table.
INSERT INTO shoe_customers_keyed
SELECT id, first_name, last_name, email
FROM shoe_customers;
Show the amount of cutomers in shoe_customers_keyed.
SELECT COUNT(*) as AMOUNTROWS FROM shoe_customers_keyed;
Look up one specific customer:
SELECT *
FROM shoe_customers_keyed
WHERE customer_id = 'b523f7f3-0338-4f1f-a951-a387beeb8b6a';
Compare it with all customer records for one specific customer:
SELECT *
FROM shoe_customers
WHERE id = 'b523f7f3-0338-4f1f-a951-a387beeb8b6a';
We also need to create Primary Key table for our product catalog.
Prepare a new table that will store unique products only:
CREATE TABLE shoe_products_keyed(
product_id STRING,
brand STRING,
model STRING,
sale_price INT,
rating DOUBLE,
PRIMARY KEY (product_id) NOT ENFORCED
);
Create a new Flink job to copy product data from the original table to the new table.
INSERT INTO shoe_products_keyed
SELECT id, brand, `name`, sale_price, rating
FROM shoe_products;
Check if only a single record is returned for some product.
SELECT *
FROM shoe_products_keyed
WHERE product_id = '0fd15be0-8b95-4f19-b90b-53aabf4c49df';
Now, you can finally check which jobs are still running, which jobs failed, and which stopped. Go to Flink (Preview) in environments and choose Flink Statements. Check what you can do here.
You can also use the Confluent CLI:
confluent login
confluent flink statement list --cloud aws --region eu-central-1 --environment <your env-id> --compute-pool <your pool id>
# Creation Date | Name | Statement | Compute Pool | Status | Status Detail
#--------------------------------+--------------------+--------------------------------+--------------+-----------+------------------------------------------
#...
# 2023-11-15 16:14:38 +0000 UTC | f041ae19-c932-403f | CREATE TABLE | lfcp-jvv9jq | COMPLETED | Table 'shoe_customers_keyed'
# | | shoe_customers_keyed( | | | created
# | | customer_id STRING, | | |
# | | first_name STRING, last_name | | |
# | | STRING, email STRING, | | |
# | | PRIMARY KEY (customer_id) NOT | | |
# | | ENFORCED ); | | |
# ....
# Exceptions
confluent flink statement exception list <name> --cloud aws --region eu-central-1 --environment <your env-id>
# Descriobe Statements
confluent flink statement describe <name> --cloud aws --region eu-central-1 --environment <your env-id>This is the end of Lab1, please continue with Lab2.