At Zepto our analytics run on mobile events. A customer taps something, an event service catches it, and it flows through Kafka into ClickHouse. The search team wants to know which result you clicked. The monetization team wants to know which ad you saw, in which city, for which brand. All of it real time, all of it at a scale where “just count the rows” stops being a joke and starts being a problem: roughly 16 billion events live in a single table.
There are plenty of engines that can hold that much data. Druid, Pinot, StarRocks, ClickHouse, take your pick. The hard part was never storage. The hard part was keeping the data honest.
Duplicates are not a bug, they are the weather
Our pipeline is asynchronous and it is at-least-once. That is a feature, not an accident. Kafka retries. Consumers retry. When a batch fails an insert, it lands in a dead letter topic and gets replayed later. Every one of those safety nets exists so we never lose an event, and every one of them can hand us the same event twice.
Then add the boring operational reality on top: rolling restarts, node migrations, upgrades. On one node migration in March, about 5 million events got routed to the DLQ, replayed cleanly, and produced roughly 86,000 duplicate events in the process. Nobody did anything wrong. That is just what at-least-once looks like on a Tuesday.
So the question was never “how do we stop duplicates.” You do not stop the weather. The question was “how do we make duplicates not matter.”
The base layer: ReplacingMergeTree
ClickHouse has a table engine built for exactly this, ReplacingMergeTree. You give
it a sorting key and a version column, and during background merges it keeps only
the newest row for each key.
ENGINE = ReplicatedReplacingMergeTree(..., _version)
PARTITION BY toYYYYMMDD(created_at)
ORDER BY (project_id, name, created_at, profile_id, id)The version column does the deciding:
_version DateTime64(9) DEFAULT now()Two rows with the same sorting key? The one with the larger _version wins. Our
86,000 duplicates from the migration were quietly collapsed by merges within a few
hours, and nobody woke up for it.
There is one catch, and it is the catch that trips up everyone who reaches for this
engine: merges are eventual, not immediate. Between the moment a duplicate lands
and the moment a merge eats it, a naive SELECT will happily count it twice.
The FINAL trap
ClickHouse gives you a keyword that seems to solve this: FINAL. Add it to your
query and you get the fully deduplicated view, right now, guaranteed.
We benchmarked it. On the events table over a wide date range, the same query went
from 46 seconds to 198 seconds, and it read 22 times more data to do it: 3.9 billion
rows instead of 172 million. On the profiles table, a burst of concurrent FINAL
queries blew past our per-query memory limit and OOMed the node.
FINAL is correct. FINAL is also a way to set your cluster on fire.
So we made a rule: the events table is never queried with FINAL. Which meant the dedup had to happen somewhere else.
The ORDER BY does the real work
The most important decision in the whole design is the sorting key, and we got it wrong the first time.
We started Mixpanel-style, sorting by (project_id, profile_id, created_at, name, id). It felt natural: group a user’s events together. But almost every query we
actually run filters by event name first (“how many add_to_cart events”), and
name was buried fourth in the key, so ClickHouse could not skip much.
We moved name up front: (project_id, name, created_at, profile_id, id). Same
data, different order, and suddenly the primary index could skip between 94 and 96
percent of granules on a typical query. Funnels that group by profile still work,
because profile_id is right there in the key too.
A sorting key is not paperwork. It is the query plan you are committing to before you have written a single query.
Dedup at the edges, not in the reader
With FINAL off the table, we pushed deduplication out to where it is cheap.
First, block-level dedup at ingest. ClickHouse can reject an identical block on
insert with insert_deduplicate=1, with insert_quorum=2 so replicas agree. That
catches the dumb retries for free before they ever become rows.
Second, a DLQ dedup worker. On a schedule, per shard and per partition, it does an
idempotent anti-join: take the replayed rows, keep the newest per id with argMax,
and insert only the ids that are not already there.
INSERT INTO events_replicated
SELECT argMax(col, _version), ...
FROM events_replicated_dlq
WHERE <partition>
GROUP BY id
HAVING id NOT IN (SELECT id FROM events_replicated WHERE <same partition>)argMax(col, _version) is the trick worth remembering. It is FINAL’s logic, keep
the newest version per key, but scoped to one partition and one worker run instead
of the whole table on every read. When a specific day needs a hard cleanup after a
messy reingest, a targeted OPTIMIZE ... PARTITION ... FINAL collapses just that
day, never the table.
The bug that made all of it worse
Here is the part I would put on a poster in the office.
For a while, our Kafka connector was inserting directly into the local replicated table, not through ClickHouse’s Distributed table. That sounds like a rounding error. It was not. Inserting locally scattered each user’s events randomly across all the shards. A single session of 80 to 120 events would smear itself across every node in the cluster.
Two things fell out of that. Dedup work went up about 4.6 times, because the same key now had to be reconciled on more parts in more places. And queries could not prune by shard, because a user’s data was everywhere.
The fix was one line of intent: route inserts through the Distributed table so
cityHash64(project_id, profile_id) decides the shard, and a user’s events land
together. Once they were co-located, dedup got cheaper and queries got a whole class
of pruning back.
What I would do differently
Two things, and they are the same thing said twice.
Route through Distributed from day one. Co-location is not an optimization you bolt on later, it is an assumption every other layer quietly depends on.
Dedup at ingest, not in the reader. ReplacingMergeTree is a wonderful safety net and a terrible primary plan. Lean on it to survive the weather, not to answer your queries.
None of this is exotic. It is ReplacingMergeTree, a carefully chosen sorting key,
argMax at the edges, and the discipline to never reach for FINAL when the cluster
is watching. Boring, on purpose. At 16 billion rows, boring is the whole point.