Deduplicating Streams with Flink SQL

I recently found myself doing a deep dive into how Apache Flink® SQL can be used for deduplication. What I discovered is that a thorough understanding of deduplication requires quite a lot of knowledge about the Flink runtime, including event time and watermarks, state management, and changelog processing. I was surprised that exploring deduplication took me so far into the weeds, and I hope it will be instructive to share what I learned.

In what follows, the details about Flink are up-to-date as of Flink 2.3.0.

Why would anyone need to deduplicate?

Before launching into the details of how to do this, I think it’s helpful to remind ourselves why deduplication is (sometimes) needed.

Many modern frameworks, such as Flink, can produce streams with exactly-once semantics. If this is how your events are produced, then you don’t have to worry about removing spurious duplicates.

But exactly-once guarantees come with a cost attached: increased latency. For some use cases, it makes sense to settle for at-least-once semantics in exchange for increased responsiveness — and that means living with the possibility of duplicates.

There are also cases where the purpose of “deduplication” isn’t to detect and eliminate exact duplicates, but rather to remove extraneous events from a verbose event stream, or to produce a summary — more of a “decluttering” operation than a strict deduplication. More about this later.

The setup I used for my experiments

I used Confluent Cloud for most of my experiments with deduplication. (If you want to try something similar using open source Apache Flink, you will likely have to make some minor adjustments.)

I used the faker connector to generate the data. I decided to create exact duplicates, and to do so deterministically. I figured this would make it easier to see what’s going on.

Here’s the setup I used:

CREATE TABLE orders (
  `order_id` STRING NOT NULL,
  `quantity` INT NOT NULL,
  `price` DECIMAL(10,2) NOT NULL,
  `created_at` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
  WATERMARK FOR `created_at` AS `created_at`
) WITH (
  'connector' = 'faker',
  'fields.order_id.expression' = '#{Internet.UUID}',
  'fields.quantity.expression' = '#{Number.numberBetween ''100'',''10000''}',
  'fields.price.expression' = '#{Number.randomDouble ''2'', ''10'', ''500''}',
  'rows-per-second' = '1'
);

-- duplicate every order priced under 100
CREATE VIEW orders_with_dups_view AS (
  SELECT * FROM orders
  UNION ALL
  SELECT * FROM orders WHERE price < 100
);

-- store the stream with duplicates in a new table
-- backed by a kafka topic with 2 partitions
CREATE TABLE orders_with_duplicates (
  WATERMARK FOR `created_at` AS `created_at`
)
DISTRIBUTED BY (order_id) INTO 2 BUCKETS
AS (
  SELECT * FROM orders_with_dups_view
);

Applying a framework for evaluating solutions

Use case: Deduplicate this stream of orders_with_duplicates, where any duplicate is an exact duplicate.

In How I Review Flink SQL Solutions, I laid out five things I pay attention to when evaluating potential solutions:

• state
• append-only vs. updating streams
• late events
• latency
• determinism

Let’s revisit those considerations with this use case in mind, and establish specific requirements.

State and state retention

For deduplication state is necessary: recognizing a duplicate requires having stored the original event for lookup and comparison.

To be able to scale up deduplication, we need to ensure that any duplicates are processed by the same worker that processed the original event. This is, in fact, precisely how Flink scales stateful workloads — the runtime is organized around key-partitioned state. In this case, the order_id is the key, and each stream source is connected by a network shuffle that sends every order to the instance of the deduplication operator that’s handling events with that order_id:

The worst case is very expensive: if you need to ensure that no duplicates will ever occur, then every unique event must be retained forever. But for most use cases, duplicates are rare, and will typically only occur in conjunction with something restarting, i.e., while recovering from a failure. If you are confident that duplicates will only occur soon after the original event, then you can safely limit how long events are kept in Flink’s state store.

You can do this by setting an idle state timeout on the state used by deduplication.

Working purely in SQL, state TTL can be applied with a session-level configuration setting, or inside the query using a SQL hint. I tend to prefer the latter approach, since it’s more granular, and gets all of the business logic together in one place. However, looking at how the STATE_TTL hint is handled by Flink’s SQL planner, I see that this particular hint can only be applied to Join and Aggregate operations.

While it is possible to use an Aggregate for certain kinds of deduplication (discussed below), I can’t recommend it. Instead, deduplication is typically done with either the Rank or Deduplicate operators, and a STATE_TTL hint in either of these will be silently ignored.

So to handle state expiry for deduplication using Flink SQL, plan on setting either table.exec.state.ttl (for open source Apache Flink) or sql.state-ttl (on Confluent Cloud) as a session-level configuration option.

Requirement: I will use state TTL to limit state retention, based on my assumption that duplicates will only occur within some interval (e.g., 2 hours).

Append-only result vs. a result with updates

In general, when a duplicate is processed, the deduplication operator has to decide whether to

• drop the duplicate, or
• retract the previously emitted, original event, and emit an update

In this scenario, any duplicates will be exact duplicates, so there’s nothing to be gained by producing an update.

This is quite important, because adopting a solution that can produce updates will limit how deduplication can be composed with other operations. For example, you might want to use the deduplicated stream as an input to an interval join, or to an ML_PREDICT function call, neither of which will accept an updating stream.

Requirement: Append-only results.

Late events

Requirement: I have no tolerance for dropping late events.

Latency

There are two cases to consider: a job doing just deduplication, for the purpose of creating a reusable data product, vs. a job performing deduplication as the first step in a longer pipeline.

For a Flink job just doing deduplication, it will need to be configured with exactly-once guarantees, which requires checkpointing. In this situation, most of the latency will come from the delay caused by the Kafka transactions being used to commit results to the output topic, which will happen as each checkpoint is finalized. Thus the checkpointing interval will act as a lower bound on the worst-case latency.

On the other hand, if deduplication is the first step in a more complex query, the latency added by deduplication can be small, provided watermarks can either be avoided, or configured with a very short delay. (More on watermarking later, as we evaluate various solutions.)

Requirement: Ideally, whenever an event with a previously unseen order_id arrives, it will be emitted immediately. Can this be achieved?

Determinism

Requirement: Let’s aim for a fully deterministic solution, and see how close we can get.

What options does Flink SQL offer for deduplication?

If you’re familiar with solutions for removing duplicates from a table in a SQL database, a few different possibilities will come to mind.

PRIMARY KEY ENFORCED

Hypothetically, the table being produced by deduplication could have a primary key with a uniqueness constraint.

But while Flink pays attention to primary keys, it does not physically enforce uniqueness during ingestion. In Flink SQL, primary keys are always PRIMARY KEY NOT ENFORCED.

Tumbling windows

Assuming that the original and any duplicates will share the same timestamp, they will all be assigned to the same window — so this will work. But it’s not a great idea, because

• This makes deduplication a temporal operation, i.e., one that depends on watermarks. This adds latency, and late events will be dropped.

• These windows won’t emit results until they are closed, so this approach adds even more latency (based on the window duration).

Even worse is what will happen if the original and any duplicates won’t always have the same timestamp. For example, this can be the case if the timestamps are log-append timestamps assigned by the Kafka brokers. Then there’s no guarantee that an original and its duplicates will be assigned to the same window, in which case this strategy falls apart completely.

SELECT DISTINCT

For a plain append-only source, you might expect that SELECT DISTINCT would only ever produce INSERT records, but the output is actually treated as being capable of producing updates.

The details: Flink SQL uses Apache Calcite as an engine for query parsing, planning, and optimization. Calcite effectively transforms SELECT DISTINCT id FROM t into SELECT id FROM t GROUP BY id; DISTINCT is nothing more than syntactic sugar. Flink’s planner treats this like any other non-windowed aggregation: GroupAggregate is unconditionally declared as update-producing.

In the case where state TTL is enabled (state expiry is needed on an unbounded DISTINCT to bound state size), a repeated key forces the operator to re-emit UPDATE_BEFORE/UPDATE_AFTER for a value that hasn’t logically changed, purely to keep the key “alive” in any downstream stateful operator’s TTL clock.

Another issue is that this approach isn’t very flexible. It works for this particular situation (with exact duplicates), but can’t be extended to cover cases where you would want more control.

E.g., consider deduplicating a stream of orders based on their order_id, where “duplicates” may vary in price, and time. The business requirement might be that the price included in the result should be the price of the last version of the order.

This is an example of decluttering, or deduplication-as-summarization, as mentioned in the intro. Deduplication based on OVER aggregations will have the flexibility to handle use cases like this.

OVER aggregation

OVER is a SQL operation that groups together each row with some or all of the previous rows that match. The effect is to transform each incoming row into an aggregation over that group of matching rows.

The query below is based on the example in the documentation. It sets up partitioning that groups together each distinct order with any duplicates, based on the order_id:

SELECT
  order_id,
  quantity,
  price,
  created_at
FROM (
  SELECT *, ROW_NUMBER()
    OVER (PARTITION BY order_id
          ORDER BY created_at ASC) AS row_num
  FROM orders_with_duplicates
)
WHERE row_num = 1;

Conceptually, how does this work?

Before looking at how the Flink SQL runtime actually executes this query, let’s try to understand why it makes sense to model deduplication this way.

PARTITION BY order_id

is effectively saying that we want to deduplicate solely on the basis of the order_id column — the orders_with_duplicates table is being sliced into disjoint partitions, where each partition contains all of orders with a specific order_id.

ORDER BY created_at ASC

Then within each partition, the order events with the same order_id are to be sorted by their created_at timestamp in ascending order (from oldest to most recent).

The inner subquery (below) is including in the result a column named row_num that indicates that row’s position within the sorted partition for that order_id.

SELECT *, ROW_NUMBER()
  OVER (PARTITION BY order_id ORDER BY created_at ASC) AS row_num
FROM orders_with_duplicates

For each distinct order_id, the row_num will have the value 1 for the first event, and then it will continue counting upwards — 2, 3, 4, etc. — for any duplicates. Deduplication then becomes a matter of filtering the results of this subquery, to keep only the first event:

SELECT order_id, quantity, price, created_at
FROM (
  SELECT *, ROW_NUMBER()
    OVER (PARTITION BY order_id ORDER BY created_at ASC) AS row_num
  FROM orders_with_duplicates
)
WHERE row_num = 1;

Variations

ORDER BY event_time ASC vs. DESC

Regardless of whether the ORDER BY specifies ASC or DESC, the runtime always sorts the input stream in ascending order.

Rather than affecting the sort itself, the direction (ASC vs. DESC) determines how duplicates are processed:

• ASC: the output is append-only, using the earliest event in each group (duplicates are dropped)
• DESC: the output is updated whenever later duplicates are processed

In general, this sorting step is necessary to guarantee the results are deterministic, but it does introduce some latency (because each record must wait to be emitted until the watermark has caught up to that record’s timestamp), and it introduces the possibility of late events and incomplete results.

The watermark interval can be adjusted to manage the tradeoff between latency and completeness:

• larger watermark interval: more latency, less risk of dropping late events
• shorter watermark interval: less latency, more risk of dropping late events

ORDER BY processing_time

This is similar to the event_time case described above, except that the sorting step isn’t done. So unless the original and any duplicates are identical, the results are non-deterministic.

The sort direction determines how duplicates are processed:

• ASC: the output is append-only, using the first event processed in each group
• DESC: the output is updated whenever duplicates are processed

Watermarks are irrelevant; lateness and latency aren’t an issue.

ORDER BY not_a_time_attribute

Similar to the processing time variant, no sorting is actually performed.

This is a good example of the idea that a SQL statement is a declarative expression of an intended outcome, rather than a prescription of how to achieve that outcome. Although this query uses ORDER BY, the Flink SQL runtime is not doing any sorting. Instead, the runtime is paying attention to whether the ORDER BY is specified as ascending or descending, and using that directional information to determine how duplicates are processed:

• ASC: the result is updated whenever a smaller duplicate is processed
• DESC: the result is updated whenever a larger duplicate is processed

While researching this blog post, I found a bug in the implementation of this particular variant. Currently, the SQL planner treats the output of the ascending case as append-only, but that’s not correct (both ASC and DESC produce streams that may contain updates). You can watch FLINK-40368 for progress on this issue.

Because the timestamps aren’t being used, watermarks and lateness aren’t issues for this approach, and the results are emitted immediately.

Summary of the pure SQL solutions

None of the solutions considered above satisfy every requirement, but a couple of them come pretty close.

If I’m being precise, then any solution will be non-deterministic, assuming state TTL is used.

OVER aggregation based on processing time. In general, this approach is non-deterministic, but for this specific case, where duplicates are identical to the original, it is deterministic (or almost so, if state TTL is used). However, Confluent Cloud doesn’t support processing time, so I’d like to find another approach.

OVER aggregation not based on time at all. The main issue here is that the output is, theoretically, an updating stream. I say theoretically because the planner is currently buggy and doesn’t correctly categorize this query; it considers the not_a_time_attribute case append-only, but in general, that’s wrong.

On the other hand, we can construct our query so it doesn’t actually produce any updates at all:

SELECT * FROM TO_CHANGELOG((SELECT order_id, quantity, price FROM
  (SELECT *, ROW_NUMBER()
    OVER (PARTITION BY order_id
          ORDER BY order_id ASC) AS row_num
  FROM orders_with_duplicates)
WHERE row_num = 1));

This has the desired effect because of ORDER BY order_id ASC. This is being applied to an aggregation group that only contains rows with the same order_id, so the runtime will never process a duplicate that would necessitate an update. (You might wonder if the “emit an update solely to keep keyed state alive” behavior we saw earlier with SELECT DISTINCT can occur here as well, but no, the Top-N / Rank operators used here emit updates only when rank/order actually changes.)

TO_CHANGELOG is used to convert the output of the deduplication from theoretically updating to a stream that is truly append-only (in the eyes of the SQL planner).

For more about TO_CHANGELOG, see the blog post by Gustavo de Morais on Flink SQL Evolution: Handling Custom CDC with FROM_CHANGELOG and TO_CHANGELOG.

What about using a custom ProcessTableFunction (PTF)?

Given the challenges we’ve seen so far, this feels like a situation where implementing a custom SQL operation, in the form of a ProcessTableFunction (PTF), might be a better approach.

The Java implementation of such a PTF turns out to be pretty small — state per key, and a check before emitting:

/**
 * Keeps the first row seen for each partition key and drops
 * any exact duplicates that follow, in arrival order. State
 * for a key expires 2 hours after that key was last seen.
 */
public class ExactDeduplication extends ProcessTableFunction<Row> {

  public static class SeenState {
    public boolean seen = false;
  }

  public void eval(
      @StateHint(ttl = "2 hours") SeenState state,
      @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
    if (!state.seen) {
      state.seen = true;
      collect(input);
    }
  }
}

Registering and calling it looks like this:

CREATE FUNCTION EXACT_DEDUPLICATION
    AS 'com.confluent.developer.ptf.ExactDeduplication';

SELECT * FROM EXACT_DEDUPLICATION(
  input => TABLE orders_with_duplicates PARTITION BY order_id,
  uid   => 'exact-dedup-orders'
);

TABLE orders_with_duplicates PARTITION BY order_id specifies that deduplication (and the key-partitioned state) is based on the order_id. The uid assigns a stable, unique identifier to stateful operations so that Flink can persist and correctly map state across job restarts.

Of course, you might prefer to implement a more broadly applicable PTF, instead of focusing on the rather narrow scenario discussed here.

Conclusion

Going back to the requirements I set out earlier — bounded state retention, append-only output, no dropped late events, immediate emission for new keys, and determinism — the OVER aggregation ordered by a non-time column, wrapped in TO_CHANGELOG, comes closest of all the pure-SQL options. However, it depends on a currently mis-categorized (and, until FLINK-40368 is fixed, arguably buggy) code path in the planner.

There’s an underlying tension at work here: Flink SQL is declarative, but the requirements I care about — such as state growth, changelog shape, and latency — are runtime properties that the SQL text doesn’t state directly. Two queries that look almost identical (ORDER BY created_at ASC vs. ORDER BY order_id ASC) are implemented by different operators (StreamExecDeduplicate vs. StreamExecRank) and differ in latency, and whether the result is append-only. The only way to know for sure is to understand what the planner does with each one.

The custom PTF sidesteps all of that. It trades the cleverness of finding the right SQL incantation for a small amount of Java, yielding output and state semantics that are exactly what I asked for, and nothing to double check against planner internals. For a narrow, well-understood use case like this one, that tradeoff is worth considering.

Deduplication looks like a solved problem, but picking a good solution in Flink SQL means understanding state, changelogs, and watermarks well enough to know what a given query is really going to do — not just what it says.