<a id="flink-sql-deduplicate-topic-action"></a>

# Deduplicate Rows in a Table with Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables generating a table that contains only unique records from
an input table with only a few clicks.

In this guide, you create a Flink table and apply the Deduplicate Rows
action to generate a topic that has only unique records, by using a
[deduplication statement](../reference/queries/deduplication.md#flink-sql-deduplication). The Deduplicate
Rows action creates a Flink SQL statement for you, but you do not need
any knowledge of Flink SQL to use it.

This guide shows the following steps:

- [Step 1: Create a users table](#flink-sql-deduplicate-topic-create-table)
- [Step 2: Apply the Deduplicate Topic action](#flink-sql-deduplicate-topic-apply-action)
- [Step 3: Inspect the output table](#flink-sql-deduplicate-topic-inspect-output)

## Prerequisites

You need the following prerequisites to use Flink in Confluent Cloud Console.

- Access to Confluent Cloud.

<a id="flink-sql-deduplicate-topic-create-table"></a>

## Step 1: Create a users table

Before you can deduplicate rows, you need a table with sample data that
contains duplicates. In this step, you create a simple `users` table and
populate it with mock records, some of which are duplicated intentionally.

1. Log in to Confluent Cloud and navigate to your Flink workspace.
2. Run the following statement to create a `users` table.
   ```sql
   CREATE TABLE users (
     user_id STRING NOT NULL,
     registertime BIGINT,
     gender STRING,
     regionid STRING
   );
   ```
3. Insert rows with mock data into the `users` table.
   ```sql
   INSERT INTO users VALUES
     ('Thomas A. Anderson', 1677260724, 'male', 'Region_4'),
     ('Thomas A. Anderson', 1677260724, 'male', 'Region_4'),
     ('Trinity', 1677260733, 'female', 'Region_4'),
     ('Trinity', 1677260733, 'female', 'Region_4'),
     ('Morpheus', 1677260742, 'male', 'Region_8'),
     ('Morpheus', 1677260742, 'male', 'Region_8'),
     ('Dozer', 1677260823, 'male', 'Region_1'),
     ('Agent Smith', 1677260955, 'male', 'Region_0'),
     ('Persephone', 1677260901, 'female', 'Region_2'),
     ('Niobe', 1677260921, 'female', 'Region_3'),
     ('Niobe', 1677260921, 'female', 'Region_3'),
     ('Niobe', 1677260921, 'female', 'Region_3'),
     ('Zee', 1677260922, 'female', 'Region_5');
   ```
4. Inspect the inserted rows.
   ```sql
   SELECT * FROM users;
   ```

   Your output should resemble:
   ```none
   user_id            registertime gender regionid
   Thomas A. Anderson 1677260724   male   Region_4
   Thomas A. Anderson 1677260724   male   Region_4
   Trinity            1677260733   female Region_4
   Trinity            1677260733   female Region_4
   Morpheus           1677260742   male   Region_8
   Morpheus           1677260742   male   Region_8
   Dozer              1677260823   male   Region_1
   Agent Smith        1677260955   male   Region_0
   Persephone         1677260901   female Region_2
   Niobe              1677260921   female Region_3
   Niobe              1677260921   female Region_3
   Niobe              1677260921   female Region_3
   Zee                1677260922   female Region_5
   ```

<a id="flink-sql-deduplicate-topic-apply-action"></a>

## Step 2: Apply the Deduplicate Topic action

In the previous step, you created a Flink table that had duplicate rows. In this
step, you apply the Deduplicate Topic action to create an output table that has
only unique rows.

1. In the navigation menu, click **Data portal**.
2. In the **Data portal** page, click the **Environment** dropdown menu and
   select the environment for your workspace.
3. In the **Recently created** section, find your **users** topic and click it
   to open the details pane.
4. Click **Actions**, and in the Actions list, click **Deduplicate topic** to
   open the **Deduplicate topic** dialog.
5. In the **Fields to deduplicate** dropdown, select **user_id**.

   Flink uses the deduplication field as the output message key. This means
   that the output topic’s row key can be different from the input topic’s
   row key, because the deduplication statement’s DISTRIBUTED BY clause
   determines the output topic’s key.

   For this example, the output message key is the `user_id` field.
6. In the **Compute pool** dropdown, select the compute pool you want to use.
7. (Optional) In the **Runtime configuration** section, select
   **Run with a service account** to run the deduplicate query with a
   service account principal. Use this option for production queries.

   #### NOTE
   The service account you select must have the DeveloperManage and
   DeveloperWrite roles to create topics, schemas, and run Flink statements.
   For more information, see [Grant Role-Based Access](../operate-and-deploy/flink-rbac.md#flink-rbac).
8. Click the **Show SQL** toggle to view the statement that the action
   runs.

   For this example, the deduplication query depends on the `registertime` field,
   so you must modify the generated statement to use the `registertime` field
   as the field to sort on.
9. Click **Open SQL editor** to modify the statement.

   A Flink workspace opens with the generated statement in the cell.
10. In the cell, replace `$rowtime` with `registertime` in the `ORDER BY`
    clause.
    ```sql
    CREATE TABLE `<your-environment>`.`<your-kafka-cluster>`.`users_deduplicate` (
           PRIMARY KEY (`user_id`) NOT ENFORCED
    ) DISTRIBUTED BY HASH(
           `user_id`
    ) WITH (
           'changelog.mode' = 'upsert',
           'value.format'='avro-registry',
           'key.format'='avro-registry'
    ) AS SELECT `user_id`, `registertime`, `gender`, `regionid` FROM (
           SELECT *,
                  ROW_NUMBER() OVER (PARTITION BY `user_id` ORDER BY registertime ASC) AS row_num
           FROM `<your-environment>`.`<your-kafka-cluster>`.`users`) WHERE row_num = 1;
    ```
11. Click **Run** to execute the deduplication query.

    The CREATE TABLE AS SELECT statement creates the `users_deduplicate` table
    and populates it with rows from the `users` table using a
    [deduplication query](../reference/queries/deduplication.md#flink-sql-deduplication).
12. When the **Statement status** changes to **Running**, you can query the
    `users_deduplicate` table.

<a id="flink-sql-deduplicate-topic-inspect-output"></a>

## Step 3: Inspect the output table

The statement generated by the Deduplicate Topic action created an
output table named `users_deduplicate`. In this step, you query the output
table to see the deduplicated rows.

- Run the following statement to inspect the `users_deduplicate` output
  table.
  ```sql
  SELECT * FROM users_deduplicate;
  ```

  Your output should resemble:
  ```none
  user_id            registertime gender regionid
  Thomas A. Anderson 1677260724   male   Region_4
  Trinity            1677260733   female Region_4
  Morpheus           1677260742   male   Region_8
  Dozer              1677260823   male   Region_1
  Agent Smith        1677260955   male   Region_0
  Persephone         1677260901   female Region_2
  Niobe              1677260921   female Region_3
  Zee                1677260922   female Region_5
  ```

## Related content

- Flink action: [Mask Fields in a Table](mask-fields.md#flink-sql-mask-fields-action)
- Flink action: [Transform a Topic](transform-topic.md#flink-sql-transform-topic-action)
- Flink action: [Create an Embedding](../../ai/embeddings/embedding-action.md#flink-sql-embedding-action)
- [Aggregate a Stream in a Tumbling Window](aggregate-tumbling-window.md#flink-sql-aggregate-tumbling-window)
- [Compare Current and Previous Values in a Data Stream](compare-current-and-previous-values.md#flink-sql-compare-current-and-previous-values)
- [Convert the Serialization Format of a Topic](convert-serialization-format.md#flink-sql-convert-format)

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