What is stream deduplication in Flink and why is it important

-- Flink SQL deduplication: keep first occurrence per txn_id
SELECT txn_id, account_id, amount, event_time
FROM (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY txn_id
            ORDER BY event_time ASC
        ) AS row_num
    FROM raw_transactions
)
WHERE row_num = 1;
-- Ascending order + row_num = 1 = keep-first
-- Descending order + row_num = 1 = keep-last

-- Source: Kafka topic with potential duplicates
CREATE TABLE raw_transactions (
    txn_id     STRING,
    account_id STRING,
    amount     DECIMAL(15, 2),
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'nbi-raw-transactions',
    'format'    = 'json',
    'properties.bootstrap.servers' = 'kafka.bank.internal:9092'
);

-- Deduplicate: keep first occurrence per txn_id
CREATE VIEW deduped_transactions AS
SELECT txn_id, account_id, amount, event_time
FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY txn_id ORDER BY event_time ASC) AS rn
    FROM raw_transactions
) WHERE rn = 1;

-- Use deduped view downstream
SELECT account_id, SUM(amount) AS total
FROM deduped_transactions
GROUP BY account_id;

package com.bank.flink;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class TransactionDeduplicator
        extends KeyedProcessFunction<String, Transaction, Transaction> {

    private ValueState<Boolean> seenState;
    private static final long TTL_MS = 3600000; // 1 hour

    @Override
    public void open(Configuration parameters) {
        seenState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("seen", Boolean.class));
    }

    @Override
    public void processElement(Transaction txn, Context ctx, Collector<Transaction> out)
            throws Exception {
        Boolean seen = seenState.value();
        if (seen == null || !seen) {
            // First occurrence: emit and mark as seen
            seenState.update(true);
            out.collect(txn);
            // Register timer to clean up state after TTL
            ctx.timerService().registerEventTimeTimer(
                ctx.timestamp() + TTL_MS);
        }
        // Duplicate: drop silently
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Transaction> out) {
        // TTL expired: clear state so a reused txn_id is treated as new
        seenState.clear();
    }
}

// Usage: key by transaction id, apply deduplicator
DataStream<Transaction> deduped = transactions
    .keyBy(Transaction::getTxnId)
    .process(new TransactionDeduplicator());

-- Keep-first: true deduplication (ORDER BY ASC, row_num = 1)
SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY txn_id ORDER BY event_time ASC) AS rn
    FROM raw_transactions
) WHERE rn = 1;

-- Keep-last: collapse changelog to newest value (ORDER BY DESC, row_num = 1)
SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY updated_at DESC) AS rn
    FROM account_updates
) WHERE rn = 1;

// Set state TTL in Flink SQL
SET 'table.exec.state.ttl' = '3600000'; // 1 hour

-- Defense in depth: dedup in stream + upsert sink
INSERT INTO account_totals  -- upsert sink keyed by account_id
SELECT account_id, SUM(amount)
FROM deduped_transactions   -- already deduplicated
GROUP BY account_id;
-- Even if a duplicate slips past the TTL, the upsert sink
-- applies it idempotently (overwrites, not double-counts)

// Senior tip: monitor dedup state size and hit rate
// Flink metrics: number of keys in dedup state
// If growing unbounded → TTL not configured or too long
// If hit rate is high → investigate why source sends so many duplicates