<!-- .. include:: ../../includes/batch-streaming-labels.rst
:start-after: batch-streaming-labels_start
:end-before: batch-streaming-labels_end -->

<a id="flink-sql-deduplication"></a>

# 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

<!-- .. include:: ../../includes/batch-streaming-labels.rst
:start-after: execution-mode-batch-and-streaming_start
:end-before: execution-mode-batch-and-streaming_end -->

The following is the general syntax for a deduplication query:

```sql
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](../../concepts/timely-stream-processing.md#flink-sql-time-attributes). Watermarks aren’t
  required. Ordering by `ASC` means keeping the record with the smallest
  value. Ordering by `DESC` means keeping the record with the largest
  value.
- `WHERE rownum = 1`: The `rownum = 1` is 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](topn.md#flink-sql-top-n). 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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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.

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

```mysql
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:

```none
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
```

## Related content

- Flink action:
  [Deduplicate Rows in a Table](../../how-to-guides/deduplicate-rows.md#flink-sql-deduplicate-topic-action)
- [Flink SQL Queries](overview.md#flink-sql-queries)
- [Flink SQL Functions](../functions/overview.md#flink-sql-functions-overview)
- [Statements](../statements/overview.md#flink-sql-statements-overview)
- [Top-N Queries](topn.md#flink-sql-top-n)
- [Window Deduplication Queries](window-deduplication.md#flink-sql-window-deduplication)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).