Deduplication Queries in Confluent Cloud for Apache Flink
Confluent Cloud for Apache Flink® enables removing duplicate rows over a set of columns in a Flink SQL table.
Syntax
The following is the general syntax for a deduplication query:
SELECT [column_list]
FROM (
SELECT [column_list],
ROW_NUMBER() OVER ([PARTITION BY column1[, column2...]]
ORDER BY order_column [asc|desc]) AS rownum
FROM table_name)
WHERE rownum = 1
Parameter specification
Note
This query pattern must be followed exactly. Otherwise, the optimizer can’t translate the query.
ROW_NUMBER(): Assigns a unique, sequential number to each row, starting with one.PARTITION BY column1[, column2...]: Specifies the partition columns by the deduplication key.ORDER BY order_column [asc|desc]: Specifies the ordering column. Any comparable column can be used. The ordering column isn’t required to be a time attribute. Watermarks aren’t required. Ordering byASCmeans keeping the record with the smallest value. Ordering byDESCmeans keeping the record with the largest value.WHERE rownum = 1: Therownum = 1is required for Flink SQL to recognize that the query is deduplication.
Description
Deduplication removes duplicate rows over a set of columns, keeping only the first or last row.
Flink SQL uses the ROW_NUMBER() function to remove duplicates, similar
to its usage in Top-N Queries in Confluent Cloud for Apache Flink. Deduplication represents a special
case of the Top-N query, in which N is one and rows are ordered by the
specified ordering column.
An upstream extract, transform, load (ETL) job that isn’t end-to-end
exactly-once can produce duplicate records in the sink during failover.
Duplicate records affect the correctness of downstream analytical jobs,
such as SUM and COUNT, so deduplication is required before
further analysis can continue.
Table types. Deduplication accepts an
append-only or updating table
as input. In most cases, it produces an updating table, because a later
row can change which row is kept for a partition. It produces an
append-only table only
when all three conditions hold: the input is already append-only, the query
keeps the first row (ORDER BY ... ASC on a column that increases
monotonically, such as $rowtime), and mini-batch execution is turned
off.
See deduplication in action
Apply the Deduplicate Topic action to generate a table that contains only unique records from an input table.
Example
This example deduplicates click records by user_id, returning the
first URL each user visited. Run the following statement in the
Flink SQL shell or in a Confluent Cloud Console workspace to produce this
result. The rows are ordered by the $rowtime column, which is
the system column mapped to the Apache Kafka® record timestamp and can be
either LogAppendTime or CreateTime.
SELECT user_id, url, $rowtime
FROM (
SELECT *, $rowtime,
ROW_NUMBER() OVER (PARTITION BY user_id
ORDER BY $rowtime ASC) AS rownum
FROM `examples`.`marketplace`.`clicks`)
WHERE rownum = 1;
Your output should resemble:
user_id url $rowtime
3246 https://www.acme.com/product/upmtv 2024-04-16 08:04:47.365
4028 https://www.acme.com/product/jtahp 2024-04-16 08:04:47.367
4549 https://www.acme.com/product/ixsir 2024-04-16 08:04:47.367