Design ad click aggregation with dedupe and late events

"Aggregate ad clicks for billing. One million clicks per second, per-minute counts per campaign, no double-counting, and events arrive up to an hour late."

Step 1: clarify (4 minutes)

Is this for billing or for dashboards? The most important question, because it sets the correctness bar and most candidates never ask it.

Dashboards   Approximate is fine. Seconds of latency matter.
             HyperLogLog for unique counts is acceptable.
             Losing 0.1% of events is invisible.

Billing      Exact. Auditable. Reconcilable to the raw event log.
             Losing 0.1% of events is losing 0.1% of revenue,
             and over-counting is fraud exposure.

Assume both, because that is the real situation, and design a two-path system: an approximate fast path for the advertiser dashboard and an exact slow path for billing. Trying to serve both from one pipeline is how these systems become either too slow or too wrong.

How late is "late", and what happens after that? Mobile SDKs buffer events offline, so an hour is optimistic and a day happens. Assume a one-hour watermark for the streaming path and a 24-hour reconciliation window for billing.

What is a duplicate? Retries from the SDK, at-least-once queue redelivery, and genuine click fraud are three different things that all look like duplicate rows. Assume deduplication by a click_id generated at the SDK, plus separate fraud filtering, because conflating them means your fraud rate looks like a bug and your bugs look like fraud.

What is the aggregation key? Assume (campaign_id, ad_id, country, device_type, minute), which is roughly 5 dimensions and matters for cardinality.

Step 2: capacity math (4 minutes)

Ingest
  1,000,000 clicks/sec, ~400 bytes/event  = 400 MB/sec = 34 TB/day
  Kafka: at 100 MB/s per broker sustained, that is ~8 brokers for
  ingest alone, more for replication (x3) -> ~24 brokers minimum.
  Partitions: at ~20k msg/s/partition, 1M/sec needs >=50 partitions.
  Take 200 for headroom and consumer parallelism.

Retention
  Raw events must be retained for the reconciliation window plus an
  audit period. 34 TB/day x 30 days = ~1 PB in object storage.
  Kafka retains 3 days (100 TB); S3/Parquet holds the rest.

Dedupe state
  1M/sec x 3600 s = 3.6 billion click_ids per hour.
  At 16 bytes/key that is 58 GB PER HOUR of exact state.
  Over a 24-hour window: 1.4 TB. This is the hard part.

Aggregation cardinality
  10,000 campaigns x 50 ads x 200 countries x 4 device types
    = 400 million possible keys, but sparse: assume ~2 million
    active keys per minute.
  2M keys x 60 min x 24 h = 2.9 billion rows/day if stored per minute.
  Roll up: keep 1-minute for 7 days, 1-hour beyond that.

Output
  2M rows/minute to the serving store = ~33,000 writes/sec. Easy.

The number that drives the design: 58 GB of exact dedupe state per hour. That single figure rules out "keep a set of seen ids" and forces the two-tier approach in step 5.

Step 3: architecture, two paths

   SDK / ad server
        │  click_id (UUID generated at click time), timestamp
        ▼
   ┌─────────────┐
   │  COLLECTOR  │  validate, enrich (geo, device), no aggregation
   └──────┬──────┘
          ▼
   ┌──────────────────────────────────────┐
   │  KAFKA  topic: clicks_raw            │  partitioned by click_id
   │  200 partitions, 3x replication      │  (for dedupe locality)
   └───┬──────────────────────────┬───────┘
       │                          │
       │ FAST PATH                │ SLOW PATH
       ▼                          ▼
  ┌─────────────┐          ┌──────────────┐
  │ Flink       │          │  S3 / Parquet │  raw events, partitioned
  │ 1-min       │          │  by event hour│  by EVENT time
  │ tumbling    │          └───────┬───────┘
  │ windows     │                  ▼
  │ watermark   │          ┌──────────────┐
  │ = 1 hour    │          │  BATCH JOB    │  runs T+2h and T+24h
  └──────┬──────┘          │  exact dedupe │
         │                 │  exact counts │
         ▼                 └───────┬───────┘
  ┌─────────────┐                  ▼
  │ dashboard   │          ┌──────────────┐
  │ store       │          │  BILLING      │  the system of record
  │ (approx)    │          │  (exact)      │
  └─────────────┘          └──────────────┘

The two paths read the same Kafka topic and produce different guarantees. The fast path is for humans looking at graphs; the slow path is for invoices. Reconciliation between them is a monitored metric, and a divergence above a threshold is an alert, because that divergence is the earliest signal that either path has a bug.

Partition by click_id, not by campaign_id. Partitioning by campaign creates a hot partition for the largest advertiser and it is the obvious-looking choice that fails. Partitioning by click_id distributes uniformly and gives dedupe locality: all occurrences of a given click land on one partition, so dedupe state is local rather than global.

Step 4: watermarks and the late-event problem

The core of the streaming path, and it is where Flink watermarks becomes concrete.

// Event time, not processing time. Billing must not depend on when
// a packet happened to arrive.
WatermarkStrategy<Click> strategy = WatermarkStrategy
    .<Click>forBoundedOutOfOrderness(Duration.ofHours(1))
    .withTimestampAssigner((click, ts) -> click.eventTimeMs)
    // Without idleness, one quiet partition stalls the global
    // watermark and every window everywhere stops firing.
    .withIdleness(Duration.ofMinutes(1));

DataStream<ClickAgg> agg = clicks
    .assignTimestampsAndWatermarks(strategy)
    .keyBy(c -> c.aggKey())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    // Allowed lateness: the window stays in state past the watermark
    // and re-fires with an updated result for stragglers.
    .allowedLateness(Time.hours(1))
    // Beyond that, do not drop silently: route to a side output that
    // the batch layer picks up. Silent drops are lost revenue.
    .sideOutputLateData(LATE_TAG)
    .aggregate(new ClickCounter());

The three-tier lateness policy is the design, and stating it as three tiers is what demonstrates understanding:

Within the watermark (1 h)     Included in the normal window firing.
Within allowed lateness (1 h)  Window re-fires with an updated count.
                               Downstream must handle a RETRACTION or
                               an upsert, not an append.
Beyond that                    Side output -> object storage -> the
                               batch layer reconciles at T+24h.
                               NEVER silently dropped.

The cost of a one-hour watermark, stated honestly: every window holds state for at least an hour past its end, so at 2 million keys per minute with an hour of retained windows that is 120 million keys in RocksDB state. That is real memory and disk on the Flink cluster, and it is why the watermark cannot simply be set to 24 hours "to be safe". The watermark is a memory-versus-completeness trade, and choosing one hour is choosing to push the remainder to the batch layer.

withIdleness is the detail that separates people who have run Flink from people who have read about it. With 200 partitions, one partition receiving no data holds the global watermark back, because the watermark is the minimum across all sources. Every window everywhere stops firing, and the symptom is "the dashboard froze but the job is healthy".

Step 5: deduplication at 58 GB per hour

Exact dedupe over 3.6 billion ids per hour is the hardest constraint. Two tiers.

Tier 1: streaming, probabilistic, cheap.

// Per-key state in Flink, scoped to the partition, with a TTL.
// Partitioning by click_id means this state is LOCAL: a given
// click_id only ever appears on one partition.
public class Deduper extends KeyedProcessFunction<String, Click, Click> {
    private transient ValueState<Boolean> seen;

    @Override
    public void open(Configuration cfg) {
        StateTtlConfig ttl = StateTtlConfig
            .newBuilder(Time.hours(2))          // 2h > 1h watermark
            .setUpdateType(OnCreateAndWrite)
            .cleanupInRocksdbCompactFilter(1000)
            .build();
        ValueStateDescriptor<Boolean> d =
            new ValueStateDescriptor<>("seen", Boolean.class);
        d.enableTimeToLive(ttl);
        seen = getRuntimeContext().getState(d);
    }

    @Override
    public void processElement(Click c, Context ctx, Collector<Click> out)
            throws Exception {
        if (seen.value() != null) return;       // duplicate, drop
        seen.update(true);
        out.collect(c);
    }
}

State size: 3.6 billion keys per hour, 2-hour TTL, so roughly 7.2 billion keys at ~20 bytes in RocksDB = 144 GB spread across the parallelism. At 200 subtasks that is under a gigabyte each, which is fine on local SSD. This is exactly why partitioning by click_id matters: the state is partitioned with it.

Tier 2: batch, exact, authoritative.

-- Runs at T+2h and again at T+24h over raw Parquet in object storage.
-- Exact dedupe across the FULL window, including events the streaming
-- layer never saw because they arrived beyond allowed lateness.
WITH deduped AS (
  SELECT click_id,
         -- Deterministic: always keep the earliest occurrence, so
         -- reruns produce identical results. Non-determinism here
         -- means the invoice changes between runs.
         MIN_BY(STRUCT(campaign_id, ad_id, country, device_type,
                       event_time), ingest_time) AS c
  FROM clicks_raw
  WHERE event_hour BETWEEN $start AND $end
  GROUP BY click_id
)
SELECT c.campaign_id, c.ad_id, c.country, c.device_type,
       DATE_TRUNC('minute', c.event_time) AS minute,
       COUNT(*) AS clicks
FROM deduped
GROUP BY 1, 2, 3, 4, 5;

Determinism is the property that matters for billing. MIN_BY(..., ingest_time) means a rerun produces byte-identical output. A non-deterministic tie-break (ANY_VALUE, or ordering by something not unique) means the invoice changes when you rerun the job, which is the kind of thing an auditor finds and you cannot explain.

And the reconciliation metric:

divergence = |streaming_count - batch_count| / batch_count

Expected: 0.1% to 0.5%, from events beyond allowed lateness.
Alert above 2%: something is wrong in one of the two paths.

That metric is the system's own self-check, and having one is a stronger answer than any individual mechanism, because it catches the failures you did not anticipate.

Step 6: exactly-once, and what it does and does not cover

Kafka -> Flink -> Kafka
  Flink's two-phase commit sink plus Kafka transactions gives
  end-to-end exactly-once WITHIN this boundary. Real, and it works.

Flink -> external database
  Only exactly-once if the sink is idempotent or transactional.
  For the aggregate store, use an UPSERT keyed by
  (agg_key, window_start), so a replay overwrites rather than
  double-counts. This is the practical answer.

Flink checkpoint -> replay
  On failure, Flink rewinds to the last checkpoint and reprocesses.
  Without an idempotent sink, that reprocessing DOUBLE COUNTS,
  which is precisely the failure this system exists to prevent.

The upsert is what makes replay safe, and it is worth stating explicitly rather than saying "exactly-once" and moving on:

INSERT INTO click_agg (campaign_id, ad_id, country, device_type,
                       window_start, clicks)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (campaign_id, ad_id, country, device_type, window_start)
DO UPDATE SET clicks = EXCLUDED.clicks;   -- SET, not ADD.

SET rather than clicks = click_agg.clicks + EXCLUDED.clicks. The window emits a complete count for that window, so overwriting is correct and adding double-counts on replay. This one-character-class distinction is the most common bug in these pipelines.

Step 7: failure modes

Flink job fails and restarts
  -> Rewinds to the last checkpoint, reprocesses, upserts overwrite.
     No double count. Dashboard may briefly show a stale value.

Kafka partition skew
  -> Partitioning by click_id (a UUID) is uniform by construction.
     Partitioning by campaign_id would create a hot partition for the
     biggest advertiser, which is the obvious-looking mistake.

One partition goes idle
  -> withIdleness prevents the global watermark stalling. Without it,
     ALL windows stop firing and the job looks healthy while the
     output is frozen. Highest-value single configuration line.

Clock skew on SDK devices
  -> Client timestamps are not trustworthy. Record BOTH client
     event_time and server ingest_time; use client time for
     windowing, and reject or clamp events whose client time is
     implausible relative to ingest time (future-dated, or older
     than the retention window).

Massive late burst (a region reconnecting)
  -> Beyond allowed lateness, so it goes to the side output and the
     batch layer picks it up. The dashboard under-reports briefly;
     billing is correct. This is the designed behaviour and it is
     worth naming as a designed behaviour rather than a failure.

Duplicate click_ids from a buggy SDK release
  -> Dedupe absorbs them, and the DEDUPE RATE is a monitored metric.
     A jump from 0.3% to 12% is an SDK bug, and you want to know
     within minutes rather than at month end.

Click fraud
  -> Explicitly NOT the dedupe layer's job. A separate scoring
     pipeline on the raw stream, with its own model and its own
     adjustment applied at billing time. Conflating them means
     fraud looks like a bug and bugs look like fraud.

Step 8: what changes at ten times the scale

At 10 million clicks per second:

Kafka ingest becomes the dominant cost. 4 GB/sec at 3x replication is 12 GB/sec of network and disk. The move is edge pre-aggregation: collectors in each region emit partial counts per (key, second) alongside a sampled raw stream, so the aggregate path carries orders of magnitude less volume while raw events still land in object storage for billing.

Exact dedupe stops being affordable in the streaming layer. 36 billion ids per hour is 720 GB of RocksDB state at the 2-hour TTL. The move is a Cuckoo filter in the fast path, accepting a small false-positive rate (which under-counts slightly on the dashboard), with exact dedupe remaining in the batch layer where billing is decided.

The batch layer's window shrinks. At this volume a 24-hour reconciliation reads a petabyte. The move is hourly incremental reconciliation with partitioned, idempotent outputs, so each hour is finalised independently and a rerun touches one hour.

Aggregate cardinality forces pre-rollup. 20 million active keys per minute at 1-minute granularity is unsustainable in a serving store; the answer is a rollup hierarchy (minute for 24 hours, hour for 30 days, day beyond) with the finest grain retained only where advertisers actually query it.

Production evidence

The Lambda architecture (Marz and Warren, Big Data) is the two-path speed-layer plus batch-layer structure used here, and the Kappa critique (Kreps, 2014) is the counter-argument that the batch layer is unnecessary complexity. This design is deliberately Lambda-shaped, and the justification is that billing needs exact reconciliation over a window longer than any tolerable streaming state size, which is the specific case where Kappa's argument is weakest.

Apache Flink's watermark and allowed-lateness semantics are documented precisely, and Akidau et al.'s "The Dataflow Model" (VLDB 2015) is the primary source for the event-time, watermark and trigger framework that Flink implements.

Google's Ads infrastructure publicly describes a streaming path for reporting and a separate reconciliation path for billing, which is direct evidence for the two-path design rather than a single pipeline.

Kafka's transactional producer and Flink's TwoPhaseCommitSinkFunction are the documented mechanism for end-to-end exactly-once within the Kafka-to-Kafka boundary, and the documentation is explicit that external sinks need idempotency, which is the reason for the upsert.

The IAB's click-measurement guidelines define what counts as a billable click and require deduplication and invalid-traffic filtering as separate concerns, which is the industry-standard basis for separating dedupe from fraud.

The debate

The case for streaming-only (Kappa): one pipeline, one set of code, no reconciliation between two implementations that will drift. Modern stream processors can replay from the log, so the batch layer is redundant complexity, and maintaining two implementations of the same aggregation is the exact duplication Lambda was criticised for.

The case for batch-only: simplest possible correctness story, trivially auditable, trivially rerunnable. Advertisers looking at yesterday's numbers do not need sub-second freshness, and most of the complexity here exists to serve a dashboard.

The case for the two-path design: they have genuinely different requirements. Dashboards need seconds and tolerate 0.5 percent error; billing needs exactness over a 24-hour window and tolerates hours of latency. One pipeline serving both is either too slow for the dashboard or holding 1.4 TB of streaming state to be exact.

My position: two paths, with reconciliation as a monitored metric and the batch layer as the system of record for billing. The Kappa objection is real and I would answer it directly: the duplication is justified here because exact dedupe over 24 hours is 1.4 TB of state, and holding that in a streaming job to avoid a batch job is a worse trade than running both. Where the window is short enough for streaming state to be affordable, I would agree with Kappa and run one path.

The decision I would defend hardest is partitioning by click_id rather than campaign_id. It looks wrong at first, because the aggregation is by campaign, and it is right for two reasons: campaign partitioning creates a hot partition for the largest advertiser, and click_id partitioning makes dedupe state local instead of global. The aggregation then does a keyBy shuffle, which costs a network hop and is far cheaper than either alternative.

The second is ON CONFLICT DO UPDATE SET rather than +=. The window emits a complete count, so overwriting is correct and adding double-counts on every checkpoint replay. That is a one-line difference that turns "exactly-once" into over-billing, and it is the most common real bug in these pipelines.

And the thing I would insist on regardless of architecture: never drop a late event silently. Beyond allowed lateness it goes to a side output and into the batch layer. A dropped click is lost revenue for the platform or an unbilled impression for the advertiser, and a pipeline that discards them quietly has no way to discover it is doing so.

Follow-up Q&A

"How do you handle events that arrive an hour late?" A three-tier policy. Within the watermark, an hour, they are included in the normal window firing. Within allowed lateness, another hour, the window stays in state and re-fires with an updated count, which means the downstream sink has to handle an upsert rather than an append. Beyond that they go to a side output, into object storage, and the batch reconciliation at T+24h picks them up. What I would never do is drop them silently, because a dropped click is lost revenue and there is no way to notice.

"Why not just set the watermark to 24 hours?" Because the watermark is a memory-versus-completeness trade. Every window holds state until the watermark passes it, so at two million keys per minute a 24-hour watermark means holding roughly 2.9 billion window keys in RocksDB. One hour is about 120 million, which is affordable. Setting it long "to be safe" moves the cost from the batch layer, where it is cheap, to the streaming layer, where it is not.

"Why partition by click_id when you're aggregating by campaign?" Two reasons. Partitioning by campaign creates a hot partition for the largest advertiser, and ad spend is extremely skewed, so that is not a theoretical concern. And partitioning by click_id makes dedupe state local: every occurrence of a given click lands on the same partition, so the dedupe check is a local state lookup rather than a distributed one. The aggregation then does a keyBy shuffle, which is one network hop and much cheaper than either alternative.

"58 GB of dedupe state per hour. How do you afford it?" Two tiers. In the streaming path, keyed state with a two-hour TTL, which at 20 bytes per key is about 144 GB spread across 200 subtasks, so under a gigabyte each on local SSD. That is affordable precisely because partitioning by click_id partitions the state with it. Exact dedupe over the full 24-hour window happens in the batch layer over Parquet in object storage, where scanning a terabyte is a normal thing to do and holding it in memory is not.

"What makes the batch job safe to rerun?" Determinism. MIN_BY(..., ingest_time) always keeps the earliest occurrence of a click_id, so a rerun produces byte-identical output. If the tie-break were ANY_VALUE or an ordering that is not unique, the invoice would change between runs, which is exactly the kind of thing an auditor finds and you cannot explain. Determinism in the dedupe tie-break is a billing requirement, not a nicety.

"You said exactly-once. What does that actually cover?" Kafka to Flink to Kafka, via Flink's two-phase commit sink and Kafka transactions, genuinely. It does not cover the external database sink, which is why the aggregate store uses an upsert keyed by aggregation key and window start. And it has to be SET, not +=: the window emits a complete count for that window, so overwriting is correct and adding double-counts on every checkpoint replay. That distinction is the most common bug in these pipelines.

"One Kafka partition stops receiving data. What happens?" Without withIdleness, the global watermark stalls, because the watermark is the minimum across all sources. Every window everywhere stops firing, and the job reports healthy while the output is frozen. It is the single highest-value configuration line in the whole job, and the symptom is "the dashboard stopped updating but nothing is failing", which is very hard to diagnose if you do not already know about it.

"How do you separate duplicates from click fraud?" Deliberately, into different systems. Dedupe is mechanical: same click_id, drop it. Fraud is a scoring problem with a model, running on the raw stream, producing adjustments applied at billing time. Conflating them is bad in both directions: a fraud spike looks like a pipeline bug, and a pipeline bug looks like fraud. And I would monitor the dedupe rate itself, because a jump from 0.3 percent to 12 percent is an SDK bug you want to hear about in minutes rather than at month end.

"Isn't this just Lambda architecture, which everyone says is bad?" Yes, and the Kappa objection is fair in general: two implementations of the same aggregation will drift. Here the duplication is justified because exact dedupe over 24 hours is 1.4 TB of state, and holding that in a streaming job purely to avoid a batch job is the worse trade. I would also monitor the divergence between the two paths as an explicit metric, expecting 0.1 to 0.5 percent and alerting above 2 percent, which turns the duplication into a self-check rather than just a risk.

Common misconceptions

"Exactly-once means you cannot double count." It covers a specific boundary. An external sink double-counts on replay unless it is idempotent, which is what the upsert is for.

"A longer watermark is safer." It is more complete and more expensive, linearly in retained window state. The batch layer is the cheaper place to buy completeness.

"Partition by the aggregation key." That creates a hot partition on skewed data and makes dedupe state global. Partition by the dedupe key and shuffle for aggregation.

"Late events can be dropped." For billing they are revenue. Route them to a side output; never discard silently.

"Dedupe handles fraud." Dedupe is mechanical and fraud is a model. Conflating them makes both undiagnosable.

Interview delivery note

Ask the billing-versus-dashboards question first, because it sets the correctness bar and most candidates skip it: "First: is this for billing or for dashboards? Because they need different things. Dashboards tolerate half a percent of error and need seconds of latency; billing needs exactness over a 24-hour window and tolerates hours. I'd design two paths reading the same topic, and monitor the divergence between them as a metric."

Then do the dedupe arithmetic, because it is what forces the design: "A million clicks a second is 3.6 billion ids an hour, which at 16 bytes is 58 gigabytes of exact dedupe state per hour. So exact dedupe over the full 24-hour window can't live in the streaming layer. Streaming gets keyed state with a two-hour TTL, batch gets exactness over Parquet in object storage."

Give the three-tier lateness policy, which is the specific thing being tested: "Within the one-hour watermark, normal window firing. Within another hour of allowed lateness, the window re-fires with an updated count, so the sink has to be an upsert rather than an append. Beyond that, a side output into object storage for the batch layer. Never a silent drop, because a dropped click is lost revenue and nothing would tell you."

Two lines that show you have operated Flink rather than read about it: "I'd partition by click_id rather than campaign_id, which looks wrong because the aggregation is by campaign. Campaign partitioning gives you a hot partition for your biggest advertiser, and click_id partitioning makes the dedupe state local." And: "and withIdleness on the watermark strategy, because with two hundred partitions one idle partition holds the global watermark back and every window everywhere stops firing while the job reports healthy."

The Staff & Team Lead Interview Reference