Flink SQL DML¶

This chapter offers continue on the best practices for implementing Flink SQL solutions, for Data Manipulation Language queries. DML is used to define statements which modify the data and don’t change the metadata.

Sources Of Information¶

• Confluent SQL documentation for DML samples
• Apache Flink SQL
• Confluent Developer Flink tutorials
• The Flink built-in system functions.

Common Patterns¶

This is important to recall that a select applies to a stream of record so the results will change at each new record. A query as below will show the last top 10 orders, and when a new record arrives this list is updated.

select * from `examples`.`marketplace`.`orders` order by $rowtime limit 10;

select count(*) AS `count` from pageviews;

set 'execution.runtime-mode' = 'batch';
# default one
set 'execution.runtime-mode' = 'streaming';

set 'sql-client.execution.result-mode' = 'table';

set 'sql-client.execution.result-mode' = 'changelog';

See this lab in changelog mode and this section in SQL concepts chapter.

Filtering¶

• Start by looking at this Confluent tutorial or The Apache Flink doc section
• SELECT ... FROM ... WHERE ... consists of column projections or filters and are stateless. Except if the output table has a a retract changelog while input is upsert, the sink will have a changelog materializer (see section below.)
• SELECT DISTINCT to remove duplicate rows. Which leads to keep state for each row.

select * from flight_events where status = 'cancelled';

select fight_id, count(*) as cancelled_fl from FlightEvents where status = 'cancelled' group by flight_id;

SELECT * FROM T1
UNION ALL
SELECT * FROM T2;

WITH filtered_data AS (
SELECT *,
    CASE 
        WHEN NOT REGEXP(user_agent, '.*Opera/.*') 
        THEN TRUE  -- Keep this row
        ELSE FALSE -- Filter out rows that match "Opera/"
    END AS keep_row
FROM examples.marketplace.clicks
)

SELECT * 
FROM filtered_data
WHERE keep_row = TRUE;

id, parent_ids, depth, information
'id_1', [], 0 , 'procedure 1'
'id_2', ['id_1'], 1 , 'operation 1'
'id_3', ['id_1','id_2'], 2 , 'phase 1'
'id_4', ['id_1','id_2','id_3'], 3 , 'phase_step 1'
'id_5', ['id_1','id_2','id_3'], 3 , 'phase_step 2'

id, procedure_id, operation_id, phase_id, phase_step_id, information
'id_1', 'id_1', NULL, NULL, NULL, 'procedure 1'
'id_2', 'id_1', 'id_2', NULL, NULL, 'operation 1'
'id_3', 'id_1', 'id_2', 'id_3', NULL, 'phase 1'
'id_4', 'id_1', 'id_2', 'id_3', 'id_4', 'phase_step 1'
'id_5', 'id_1', 'id_2', 'id_3', 'id_5', 'phase_step 2'

with `procedures` as (
    select 'id_1' as id, array[''] as parentIds, 0 as `depth` , 'procedure 1' as info
    UNION ALL
    select 'id_2' as id, array['id_1'] as parentIds, 1 as `depth` , 'operation 1' as info
    UNION ALL
    select 'id_3' as id, array['id_1','id_2'] as parentIds, 2 as `depth`, 'phase 1' as info
    UNION ALL
    select 'id_4' as id, array['id_1','id_2','id_3'] as parentIds, 3 as `depth`, 'phase_step 1' as info
    UNION ALL
    select 'id_5' as id, array['id_1','id_2','id_3'] as parentIds, 3 as `depth`, 'phase_step 2' as info
    )
select 
    id, 
    parent_id, 
    case when `depth` = 3 then id end as phase_step_id,
    case when `depth` = 2 then id end as phase_id,
    case when `depth` = 1 then id end as operation_id,
    case when `depth` = 0 then id end as procedure_id,
    info from `procedures` cross join unnest(parentIds) as ids(parent_id)

Deduplication¶

Deduplication will occur on upsert table with primary key: the last records per $rowtime or timestamp will be kept. When the source table is in append mode, the approach is to use the ROW_NUMBER() function:

SELECT ip_address, url, TO_TIMESTAMP(FROM_UNIXTIME(click_ts_raw)) as click_timestamp
FROM (
    SELECT *,
    ROW_NUMBER() OVER ( PARTITION BY ip_address ORDER BY TO_TIMESTAMP(FROM_UNIXTIME(click_ts_raw)) DESC) as rownum 
    FROM clicks
    )
WHERE rownum = 1;

See this example in the Confluent product documentation which demonstrates there is no duplicate in the Job session with select * from dedup_table; return 8 messages. Same in the topic with 8 messages too.

But it does not demonstrate the last message is kept. The deduplication sample demonstrates that an upsert table is already removing duplicates, and keep the last record per key.

Transformation¶

 TO_TIMESTAMP(FROM_UNIXTIME(click_ts_epoch)) as click_ts

TO_TIMESTAMP('2024-11-20 12:34:568Z'),

to_timestamp(transaction_date, 'yyyy-MM-dd HH:mm:ss') as tx_date, -- bigint

WHEN TIMESTAMPDIFF(day, event.event_launch_date, now()) > 120 THEN ...

 TIMESTAMPDIFF(DAY, CAST(created_date AS TIMESTAMP_LTZ(3)), CURRENT_DATE) as days_since_launch,

 CREATE TABLE my_table (
    key_col INT,
    nested_data ARRAY<ROW<id INT, name STRING>>
) WITH (...)

 SELECT
    t.key_col,
    unnested_row.id,
    unnested_row.name
FROM
    my_table AS t
CROSS JOIN UNNEST(t.nested_data) AS unnested_row;

SELECT array_field[4] FROM ((VALUES ARRAY[5,4,3,2,1])) AS T(array_field)

CREATE VIEW visited_pages_per_minute AS 
SELECT 
    window_time,
    user_id, 
    ARRAY_AGG(url) AS urls
FROM TABLE(TUMBLE(TABLE `examples.marketplace.clicks`, DESCRIPTOR(`$rowtime`), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, window_time, user_id;
-- once the view is created
SELECT * from visited_pages_per_minute;

-- it is possible to expand an array into multiple rows using cross join unnest

SELECT v.window_time, v.user_id, u.url FROM visited_pages_per_minute AS v
CROSS JOIN UNNEST(v.urls) AS u(url)

create table users_msk like users;
INSERT INTO users_msk SELECT ..., REGEXP_REPLACE(credit_card,'(\w)','*') as credit_card FROM users;

Statement Set¶

The benefit of bundling statements in a single set is to reduce the repeated read from the source for each Insert. A single read from the source is executed and shared with all downstream INSERTS.

Do not use Statement Set when the source are different for all statements within the Statement Sets. Take into account that within the statement set if one statement fails, then all queries fail. The state is shared by all the statements within Statement set, so one stateful query can impact all other statements.

    create table late_orders
    with (
        'connector'= ''
    ) 
    LIKE orders (EXCLUDING OPTIONS)

EXECUTE STATEMENT SET
BEGIN
    INSERT INTO late_orders SELECT from orders WHERE `$rowtime` < CURRENT_WATERMARK(`$rowtime`);
    INSERT INTO order_counts -- the sink table
        SELECT window_time, COUNT(*) as cnt
        FROM TABLE(TUMBLE(TABLE orders DESCRIPTOR(`$rowtime`), INTERVAL '1' MINUTE))
        GROUP BY window_start, window_end, window_time
END

Stateful aggregations¶

An aggregate function computes a single result from multiple input rows.

Group BY¶

Classical SQL grouping of records, but with Streaming the state may grow infinitely. The size will depend of the # of groups and the amount of data to keep per group.

Distinct¶

Remove duplicate before doing the aggregation:

  ARRAY_AGG(DISTINCT user_name) as persons

OVER¶

SELECT 
    order_id,
    customer_id,
    `$rowtime`,
    SUM(price) OVER w AS total_price_ten_secs, 
    COUNT(*) OVER w AS total_orders_ten_secs
FROM `examples`.`marketplace`.`orders`
WINDOW w AS (
    PARTITION BY customer_id
    ORDER BY `$rowtime`
    RANGE BETWEEN INTERVAL '10' SECONDS PRECEDING AND CURRENT ROW
)

-- compute the total price and # of orders for a period of 10s for each customer
WITH orders_ten_secs AS ( 
SELECT 
    order_id,
    customer_id,
    `$rowtime`,
    SUM(price) OVER w AS total_price_ten_secs, 
    COUNT(*) OVER w AS total_orders_ten_secs
FROM `examples`.`marketplace`.`orders`
WINDOW w AS (
    PARTITION BY customer_id
    ORDER BY `$rowtime`
    RANGE BETWEEN INTERVAL '10' SECONDS PRECEDING AND CURRENT ROW
    )
),
-- get previous orders and current order per customer
orders_ten_secs_with_lag AS (
SELECT 
    *,
    LAG(total_price_ten_secs, 1) OVER w AS total_price_ten_secs_lag, 
    LAG(total_orders_ten_secs, 1) OVER w AS total_orders_ten_secs_lag
FROM orders_ten_secs
WINDOW w AS (
    PARTITION BY customer_id
    ORDER BY `$rowtime`
    )
-- Filter orders when the order price and number of orders were above some limits for previous or current order aggregates
)
SELECT customer_id, 'BLOCK' AS action, `$rowtime` AS updated_at 
FROM orders_ten_secs_with_lag 
WHERE 
    (total_price_ten_secs > 300 AND total_price_ten_secs_lag <= 300) OR
    (total_orders_ten_secs > 5 AND total_orders_ten_secs_lag <= 5)
UNION ALL 
SELECT customer_id, 'UNBLOCK' AS action, `$rowtime` AS updated_at 
FROM orders_ten_secs_with_lag 
WHERE 
    (total_price_ten_secs <= 300 AND total_price_ten_secs_lag > 300) OR
    (total_orders_ten_secs <= 5 AND total_orders_ten_secs_lag > 5);

 json_query(task.object_state, '$.dueDate') AS due_date,

json_query(`data`, '$' RETURNING ARRAY<STRING>) as anewcolumn

SELECT existing_column, anewcolumn from table_name
cross join unnest (json_query(`data`, '$' RETURNING ARRAY<STRING>)) as t(anewcolumn)

SELECT explode(col1) from values (array(10,20)), (null)

select map_entries(map_from_arrays())

SELECT 
    *
    FROM `stocks`
    WHERE  
    CASE 
        WHEN price > 200 THEN 'high'
        WHEN price <=200 AND price > 150 THEN 'medium'
        ELSE 'low'
    END;

ALTER TABLE <table_name> MODIFY WATERMARK for `$rowtime` as `$rowtime` - INTERVAL '20' SECONDS

Joins¶

When doing a join in a database, the result reflects the state of the join at the time we execute the query. In streaming, as both side of a join receive new rows, both side of joins need to continuously change. This is a continuous query on dynamic tables, where the engine needs to keep a lot of state: each row of each table.

This is the common join we do between two tables:

SELECT t.amount, t.order_type, s.name, s.opening_value FROM transactions t
LEFT JOIN stocks s
ON t.stockid = s.id

On the left side, the fact table, has high velocity of changes, but the events are immutables, while on the right side, the dimension, the new records arrive slowly.

When doing a join, Flink needs to fully materialize both the right and left of the join tables in state, which may cost a lot of memory, because if a row in the left-hand table (LHT), also named the probe side, is updated, the operator needs to emit an updated match for all matching rows in the right-hand table (RHT) or build side. The cardinality of right side will be mostly bounded at a given point of time, but the left side may vary a lot. A join emits matching rows to downstream operator.

The key points to keep in mind are:

• The order of joins is important, try to get the first join done on table with the lowest update frequency.
• Cross join makes the query fails
• When the RHS is an upsert table, the result will be upsert too. Which means a result will be re-emitted if the RHS change. To avoid that we need to take the reference data, RHS, in effect at time of the event on LHS. The result is becoming time-versioned.

  SELECT t.amount, t.order_type, s.name, s.opening_value FROM transactions t
  LEFT JOIN stocks s
  FOR SYSTEM_TIME AS OF t.purchase_ts
  ON t.stockid = s.id
  

• Temporal joins help to reduce the state size, as we need to keep recent records on both side. The time will be linked to the watermark progress.
• INNER JOIN is a cartesian product.
• OUTER joins like left, right or full, may generate records with empty columns for non-matching join.
• INTERVAL JOIN equires at least one equi-join predicate and a join condition that bounds the time on both sides.

  SELECT t.amount, t.order_type, s.name, s.opening_value FROM transactions t, stocks s
  WHERE t.stockid = s.id AND t.ts BETWEEN s.ts - INTERVAL '6' HOURS AND s.ts
  

Here is a list of important tutorials on Joins:

• Confluent Cloud: video on joins with details on fow joins work.
• Confluent -developer: How to join streams. The matching content is in flink-sql/04-joins folder for Confluent Cloud or Platform for Flink. This folder also includes more SQL exercises.
• Confluent temporal join documentation.
• Window Join Queries in Confluent Cloud for Apache Flink
• Temporal Join Study in this repo.

SELECT [column_list]
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ rowtime } [AS <alias2>]
ON table1.column-name1 = table2.column-name1

-- use separate statements to create the tables
create table Transactions (ts TIMESTAMP(3), tid BIGINT, amount INT);
create table Payments (ts TIMESTAMP(3), tid BIGINT, type STRING);
create table Matched (tid BIGINT, amount INT, type STRING);

execute statement set
begin
insert into Transactions values(now(), 10,20),(now(),11,25),(now(),12,34);
insert into Payments values(now(), 10, 'debit'),(now(),11,'debit'),(now(),12,'credit');
insert into Matched 
    select T.tid, T.amount, P.type
    from Transactions T join Payments P ON T.tid = P.tid 
    where P.ts between T.ts and T.ts + interval '1' minutes;
end

Windowing / Table Value Functions¶

Windowing Table-Valued Functions groups the Tumble, Hop, Cumulate, and Session Windows. Windows split the stream into “buckets” of finite size, over which we can implement logic. The return value adds three additional columns named “window_start”, “window_end”, “window_time” to indicate the assigned window.

• The TUMBLE function assigns each element to a window of specified window size. Tumbling windows have a fixed size and do not overlap.

SELECT window_start, product_type, count(product_type) as num_ptype
    FROM TABLE(
        TUMBLE(
            TABLE events,
            DESCRIPTOR(`$rowtime`),
            INTERVAL '10' MINUTES
        )
    )
    GROUP BY window_start, window_end, ;

INSERT INTO app_orders
select 
    window_start, 
    window_end, 
    customer_id, sum(order_amount) 
from table(tumble(table `daily_spend`, DESCRIPTOR(transaction_ts), interval '24' hours)) 
group by window_start, window_end, customer_id 

select
    flight_id,
    evt_type,
    count(evt_type) OVER w as number_evt,
from flight_events
window w as( partition by flight_id order by $rowtime rows between 100 preceding and current row);

    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;

SELECT window_start, window_end, SUM(price) as `sum`
    FROM TABLE(
        CUMULATE(TABLE `examples`.`marketplace`.`orders`, DESCRIPTOR($rowtime), INTERVAL '30' SECONDES, INTERVAL '3' MINUTES))
    GROUP BY window_start, window_end;

SinkUpsertMaterializer¶

When operating in upsert mode and processing two update events, a potential issue arises. If input operators for two tables in upsert mode are followed by a join and then a sink operator, update events might arrive at the sink out of order. If the downstream operator's implementation doesn't account for this out-of-order delivery, it can lead to incorrect results.

Flink typically determines the ordering of update history based on the primary key (or upsert keys) through a global analysis in the Flink planner. However, a mismatch can occur between the upsert keys of the join output and the primary key of the sink table. The SinkUpsertMaterializer operator addresses this mapping discrepancy.

This operator maintains a complete list of RowData in its state to correctly process any deletion events originating from the source table. However, this approach can lead to a significant state size, resulting in increased state access I/O overhead and reduced job throughput. Also the output value for each primary key is always the last (tail) element in the maintained list. It is generally advisable to avoid using SinkUpsertMaterializer whenever possible.

Consider a scenario where 1 million records need to be processed across a small set of 1,000 keys. In this case, SinkUpsertMaterializer would need to store a potentially long list, averaging approximately 1,000 records per key.

To mitigate the usage of SinkUpsertMaterializer:

• Ensure that the partition keys used for deduplication, group aggregation, etc., are identical to the sink table's primary keys.
• SinkUpsertMaterializer is unnecessary if retractions are generated using the same key as the sink table's primary key. If a large number of records are processed but most are subsequently retracted, SinkUpsertMaterializer can significantly reduce its state size.
• Utilize Time-To-Live (TTL) to limit the state size based on time.
• A higher number of distinct values per primary key directly increases the state size of the SinkUpsertMaterializer.

Row pattern recognition¶

create table StockTicker(symbol string, price int tax int) with ('connector' = 'kafka',...)
SELECT * From StockTicker 
MATCH_RECOGNIZE ( 
    partition by symbol 
    order by rowtime
    measures
        FIRST(A.rowtime) as start_tstamp,
        LAST(A.rowtime) as last_tstamp,
        AVG(A.price) as avgPrice
    ONE ROW PER MATCH
    AFTER MATCH SKIP PAST LAST ROW
    PATTERN (A+ B)
    DEFINE
        A as AVG(A.price) < 15
);

Confluent Cloud Specifics¶

See Flink Confluent Cloud queries documentation.

Each topic is automatically mapped to a table with some metadata fields added, like the watermark in the form of $rowtime field, which is mapped to the Kafka record timestamp. To see it, run describe extended table_name; With watermarking. arriving event records will be ingested roughly in order with respect to the $rowtime time attribute field.

select 'flight_id', 'aircraft_id', 'status', $rowtime
from Aircrafts
order by $rowtime;

confluent flink quickstart --name my-flink-sql --max-cfu 10 --region us-west-2 --cloud aws

ALTER TABLE table_name SET (
    'scan.startup.mode' = 'specific-offsets',
    'scan.startup.specific-offsets' = 'partition:0,offset:25; partition:1,offset:10'
);
-- Returns from offsets 26 and 11
SELECT * FROM table_name;

confluent iam service-account create my-service-account --description "new description"
confluent iam service-account list
confluent iam service-account describe <id_of_the_sa>

confluent flink statement create my-statement --sql "SELECT * FROM my-topic;" --compute-pool <compute_pool_id> --service-account sa-123456 --database my-cluster

Analyzing Statements¶

confluent environment list
confluent flink compute-pool list
confluent flink statement list --cloud aws --region us-west-2 --environment <your env-id> --compute-pool <your pool id>

Understand the physical execution plan for a SQL query¶

See the explain keyword or Confluent Flink documentation for the output explanations.

explain select ...

Indentation indicates data flow, with each operator passing results to its parent.

Review the state size, the changelog mode, the upsert key... Operators change changelog modes when different update patterns are needed, such as when moving from streaming reads to aggregations.

Pay special attention to data skew when designing your queries. If a particular key value appears much more frequently than others, it can lead to uneven processing where a single parallel instance becomes overwhelmed handling that key’s data. Consider strategies like adding additional dimensions to your keys or pre-aggregating hot keys to distribute the workload more evenly. Whenever possible, configure the primary key to be identical to the upsert key.

Troubleshooting SQL statement running slow¶

SELECT 
    id, 
    tenant_id, 
    count(*) as record_count,
FROM table_name 
GROUP BY id, tenant_id

WITH key_stats AS (
    SELECT 
        id,
        tenant_id,
        count(*) as record_count
    FROM src_aqem_tag_tag 
    GROUP BY id, tenant_id
),
distribution_stats AS (
    SELECT 
        AVG(record_count) as mean_count,
        STDDEV(record_count) as stddev_count,
        PERCENTILE_APPROX(record_count, 0.75) as q3,
        PERCENTILE_APPROX(record_count, 0.95) as p95,
        PERCENTILE_APPROX(record_count, 0.99) as p99
    FROM key_stats
)
SELECT 
    ks.*,
    ds.mean_count,
    ds.stddev_count,
    -- Z-score calculation for outlier detection
    CASE 
        WHEN ds.stddev_count > 0 
        THEN (ks.record_count - ds.mean_count) / ds.stddev_count
        ELSE 0
    END as z_score,
    -- Hot key classification
    CASE 
        WHEN ks.record_count > ds.p99 THEN 'EXTREME_HOT'
        WHEN ks.record_count > ds.p95 THEN 'VERY_HOT'
        WHEN ks.record_count > ds.q3 * 1.5 THEN 'HOT'
        ELSE 'NORMAL'
    END as hot_key_category
FROM key_stats ks
CROSS JOIN distribution_stats ds
WHERE ks.record_count > ds.mean_count 

Confluent Flink Query Profiler¶

This is a specific, modern implementation of the Flink WebUI, used to monitor the performance of the query.

• Query Profiler Product documentation
• See how to use QP to debug statement, in the techno chapter.