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.
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