Design Patterns in Stream Processing: Deduplication

I recently found myself doing a deep dive into how 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.

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.

How to compare and evaluate different approaches

When working with Apache Flink’s SQL API, it’s not unusual for several possible solutions to suggest themselves. Some solutions may have unexpected consequences, affecting flexibility, performance, cost, and maintainability.

In this section I’ll share what I’m looking for and thinking about when I review a Flink SQL statement.

Pay attention to state

Flink is a distributed framework that supports continuous, non-stop execution of stream processing pipelines. The Flink runtime and its APIs provide developers with the means to perform stateful operations on real-time data streams, while helping them manage the risk that the active state will grow until it has exhausted the underlying storage.

Different SQL operations scale differently as they process more and more records. In general, I always aim to choose operations whose state requirements grow more slowly. For example, consider the difference between these two queries:

• find all orders that were not filled within 48 hours
• find all orders that were never filled

The first query is inherently less expensive, because each order can be removed from consideration whenever (1) it is filled, or (2) after 48 hours, whichever comes first. On the other hand, the second query requires keeping track of every order that has not yet been filled, since the beginning of time.

Looking across the various operations supported by Flink SQL in streaming mode, each of them falls into one of these categories:

• stateless
• constant state
• time-bounded state
• per-key state
• per-record state

In general, time-bounded operations are safe; operations that need per-key state are safe if and only if the keyspace is either fixed or growing slowly; and operations that need to keep track of every record should be avoided whenever possible.

When evaluating potential solutions for your use cases, look for solutions that need less state, or that offer flexibility and control over that state. For example, most Flink SQL operations support a state time-to-live (TTL) configuration setting.

For more details about how much state different SQL operations require, and how to apply state TTL, see the video on using EXPLAIN for troubleshooting from the free Apache Flink SQL course on Confluent Developer.

Prefer operations that produce insert-only streams

Event streams come in different flavors:

• insert-only (or append-only) streams: e.g., a stream of clickstream events, where each click is an immutable fact
• updating streams: e.g., a stream of user records, where users come and go and their attributes can be modified

Typically, processing an insert-only stream is less expensive. The intuition that explains this is that in many situations, processing an input stream that includes updates and deletions requires maintaining a lot more state. This state is used to remember what has already been emitted, so that the effects of updates and deletions can be computed, and then reflected in the output.

Another reason to prefer insert-only streams is that some Flink SQL operations cannot (yet?) be applied to updating streams.

Of course, for some use cases, a result that includes updates is the only approach that makes sense. But don’t just blindly accept a solution that produces an updating stream; instead, ask yourself if there might be another approach.

For more details on this topic, see the video on changelog processing and troubleshooting errors about change updates.

Do you care if late events might be dropped?

In the sections above, I steered you toward Flink’s temporal operations both because they are able to limit how much state they use, and because they produce insert-only streams as output. These operations include windowed aggregations, and interval and temporal joins.

However, there is a downside to these temporal operations. In particular, they require that you either

• use processing (or wall clock) time as the basis for timing and ordering, which will make the results troublingly non-deterministic, or
• rely on timestamps in the events as the basis for timing and ordering information, along with watermarks

Because of the drawbacks involved, Confluent Cloud for Apache Flink doesn’t support operations that are based on processing time.

As for relying instead on timestamps carried by the events, it isn’t possible to know with certainty when all of the data that would affect a given operation has been ingested.

For example, consider this example, presented earlier:

find all Orders that were not filled within 48 hours

For any given Order, how long will you wait for a matching Fulfillment event? Well, however long you do wait, eventually you will need to stop waiting and report that order as being unfulfilled.

Now comes the crucial part: suppose that after an order has been deemed unfulfilled within 48 hours, a matching fulfillment event arrives, proving that the order was actually fulfilled on time.

There are only two possibilities for what the Flink runtime might do in this situation:

• If this operation has been designed to produce an insert-only result stream, and the order has already been emitted downstream, incorrectly classified as unfulfilled, then that result cannot be retracted or updated. Flink can either drop this late fulfillment event, or put it in a dead letter queue (DLQ) for some other process to deal with.

• Or, hypothetically, this operation could’ve been implemented with updates and retractions in mind. In which case, the late event could be handled. Except that doing so negates the advantage of applying the 48 hour time limit – namely, being able to clear the state for orders that remain unfulfilled.

The bottom line is this: Flink SQL’s temporal operations are not able to include the impact of late events in their results. (Lateness has a very specific technical meaning, related to watermarks, but think of it as meaning “excessively out-of-order”.)

You have some options:

• If reducing the likelihood of incomplete results is good enough for your use case, then configure a custom watermark strategy and monitor the metric that counts the late events that were dropped.

• If you must account for every event, then you will want to arrange for late events to be collected in a DLQ.

• In some (rather unusual) cases it makes sense to avoid this issue entirely by using a query that doesn’t rely on watermarks. I’ll share an example of this below, when discussing deduplication and watermarks.

How much do you care about reducing latency?

There are use cases where different solutions can have significantly different costs, in terms of the processing latency.

Many factors can add latency; these are the most significant:

• the checkpoint interval (only relevant for exactly-once semantics)
• the watermark delay (only relevant when watermarks are needed)
• timeframes built into a query, e.g., window duration

Determinism

Deterministic jobs are free of race conditions. They are easier to understand, easier to test, and less likely to require painful debugging.

Significant causes of non-determinism include:

• the use of processing time (aka wall clock time), rather than event time
• aggressively tuned watermarking that results in many late events
• making calls to non-deterministic functions, such as now()

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 on 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
);

What about deduplication?

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

Earlier, I talked about

• state
• insert-only vs. updating streams
• late data
• latency
• determinism

Before considering some specific solutions, 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:

For this use case, I’m willing to assume that in production, duplicates will only occur as a side effect of something like an outage or a restart, and thus any duplicates will follow reasonably soon after the original event.

Requirement: I’m looking for a solution that will allow me to limit state retention, based on my assumption that duplicates will only occur within some interval (e.g., 2 hours).

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

Requirement: For this scenario, I’m looking for an approach to deduplication that produces insert-only results.

Late events

Requirement: For this use case, I have no tolerance for dropping late events.

Latency

Requirement: Whenever an event with a previously unseen order_id arrives, I want it to be emitted immediately.

Determinism

Requirement: Let’s aim for a fully deterministic solution, and see if that’s possible.

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 also adds latency related to 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 duplicate will be assigned to the same window, in which case this strategy falls apart completely.

SELECT DISTINCT

One issue with SELECT DISTINCT is that Flink’s SQL planner always treats the result as an updating stream.

For a plain append-only source, in practice SELECT DISTINCT will only ever produce INSERT records, but because of some quirks in how Calcite and Flink plan this query, the output will be treated as capable of producing updates.

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 result should be 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 paritioning 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 original 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.

The sort direction 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).

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 to be 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 queries 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 timing isn’t relevant, watermarks, lateness, and latency aren’t issues for this approach.

Summary of the pure SQL solutions

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

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. However, given that Confluent Cloud doesn’t support this, I’d like to find a better approach.

OVER aggregation not based on time at all. This fails only because the output is, theoretically, an updating stream. I say theoretically because (a) the planner is currently buggy and doesn’t correctly categorize this query, and because (b) 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.

TO_CHANGELOG is used here 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).

This same technique could also be used to make SELECT DISTINCT into a viable solution for this use case.

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), would be a sensible 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.

How expensive is deduplication?

When I’m talking about Flink, I generally use expensive to mean “uses a lot of state”, or “adds considerable latency”.

Considering cost: State

The worst case can be 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. The PTF (shown above) takes care of this in the Java code.

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, based on how I see the STATE_TTL hint being handled by Flink’s SQL planner, this particular hint can only be applied to Join and Aggregate operations, and not to the Rank or Deduplicate operations used for deduplication. Query hints attached to unmatched nodes are cleared, so if you try to use state TTL here, it will be silently ignored.

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

Considering cost: Latency

Any Flink job performing deduplication will want to be configured with exactly-once guarantees, which requires checkpointing. Most of the unavoidable latency will be 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 latency you can expect.

Conclusion

Going back to the requirements I set out earlier — bounded state retention, insert-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.

This is really the tension running through the whole post: Flink SQL is declarative, but the requirements I care about — state growth, changelog shape, watermark behavior — are all 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) can differ in latency, determinism, and whether the result is append-only, and 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, in exchange for output and state semantics that are exactly what I asked for, with nothing to double check against planner internals. For a narrow, well-understood use case like this one, that trade is worth strongly considering.

If there’s one takeaway, it’s this: 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.