Keeping ClickHouse and Elasticsearch in Sync

Overview

The thing being built, five times over, is a search box over scraped TikTok data. Someone picks a follower range, a country and a minimum engagement rate, and expects twenty creators back immediately, with an exact total underneath.

Serving that from the raw tables is not close to possible. One creator card needs their last twelve videos and the statistics on each, which means joining a table of tens of millions of rows against a snapshot table of billions, once per card, on every request, for everyone searching at the same time. ClickHouse will happily do that join. It will not do it in 200 ms while forty other people are waiting.

So the data lives twice. Scrapers write raw rows to ClickHouse. A loader reads windows of those rows, reshapes them into one document per entity, and pushes them to Elasticsearch. The app only ever reads Elasticsearch, and the whole response leaves the box in 10 to 100 ms.

The rule is written down in the prune script, which is the place it matters most:

Index and product names in this piece are generalized. The shapes and the numbers are as measured.

Why two stores

Running one dataset through two engines is a real cost: two schemas, a loader between them, and a whole category of bug where the copies disagree. It is worth being precise about what that buys. These two are good at opposite questions, and the product asks both.

The row counts are the argument. A creator document is built from roughly 12 recent videos, but those videos live in a table with tens of millions of rows and their stats live in a snapshot table with billions. Doing that join at request time is the thing the split exists to avoid.

There is a control group for this, and it is the product from the top of the page. With no ClickHouse underneath it, Elasticsearch has to be the store and the serving layer at once, and the ten minute timeout on its interactive table search is what that costs. Everything in the right-hand column above it still has to do the work of the left-hand one.

Index settings

Once the index is a copy rather than the original, a whole class of settings stops being reckless. Every line in this block would be a bad idea on a primary store, and all of them are set on every serving index here:

settings: {
  index: {
    number_of_shards: 8,
    number_of_replicas: 0,
    refresh_interval: '5m',
    codec: 'best_compression',
    translog: { durability: 'async', sync_interval: '120s', flush_threshold_size: '1gb' },
    merge: { scheduler: { max_thread_count: 4 } },
    sort: { field: 'create_time', order: 'desc' },
  },
}

After a load finishes, the index is refreshed, force merged to one segment per shard, flushed, and then set to blocks.write: true with the refresh interval pushed out to an hour. A read-only index of single segments is the fastest shape Elasticsearch has.

Shard counts differ by index size: 8 for the video index, 4 for the chart indices. Neither uses aliases, which is a gap covered in failure modes.

Document shape

The two stores cannot share a schema, and the reason is easiest to see side by side. ClickHouse keeps one sound as three separate streams of rows: a hub row with its metadata, a snapshot row for every day it was measured, and a log row recording that the scraper visited. The page needs all of that as a single object.

Collapsing N snapshot rows into one nested array is the entire reason the boundary exists. The page wants a sparkline and a percentage. The record cannot produce either without a scan.

The projection

A row does not arrive in Elasticsearch looking much like it left. Four things happen to it on the way, and only the first is obvious.

Computing at load time rather than query time is what makes the page fast, and it has a cost that is worth stating plainly, because it is written into the project's own notes:

One projection step invents data. When exactly one day is missing from a series, a trend-following randomized point is generated so the chart has no hole. Roughly 0.3% of plotted points are synthetic. That is defensible for a sparkline and indefensible for anything a customer exports, which is why it is documented next to the function rather than in a commit message.

The loader also reads its own output. Cross-platform artist identity is attached by querying the already-published artist index while building the music index, which makes loader ordering a correctness constraint rather than a preference.

Windowing

The first version walked the source table by id, in order. That holds up until the table is 400 million rows and a full pass takes long enough that the rows read first are stale before the rows read last arrive. It also fails badly on a crash: you are left with the low ids updated, the high ids untouched, and no way to tell from the data which is which.

What replaced it has no cursor at all. Every entity carries a hash bucket computed with FNV-1a over its id, mapped into 1 to 1,000,000,000, and the sync walks that keyspace in fixed slices.

export function getByBatchSize(batchSize: number, maxValue: number = 1_000_000_000) {
  const ranges: [number, number][] = [];
  for (let i = 0; i < maxValue; i += batchSize) {
    ranges.push([i, Math.min(i + batchSize, maxValue)]);
  }
  return ranges;
}

Two properties come out of this. A window is a uniform random sample of entities rather than a correlated slice, so partial progress is representative. And because the bucket leads the sort key on every source table, each window is a range scan rather than a filter over a full table.

The window sizes are tuned per target, not shared:

Embedding batches are smaller on purpose. At 1152 dimensions a vector is about 4.5 KB, so 5,000 of them is a 22 MB bulk body, which is already at the top of what is comfortable.

Watermarks

Every one of these products has to answer the same question when it restarts: what did I already send? I answered it four different ways across five codebases, which is not a flattering thing to discover, and the differences matter more than I expected them to.

The sharded variant is the one to copy. Pushing the exclusion into the same window the outer query already scans keeps the anti-join inside one shard instead of reading the whole tracker:

SELECT content_id, author_batch_id, embedding
FROM ThumbnailEmbed FINAL
WHERE author_batch_id >= ${start} AND author_batch_id < ${end}
  AND content_id NOT IN (
    SELECT content_id FROM ContentEmbedElasticUpload FINAL
    WHERE author_batch_id >= ${start} AND author_batch_id < ${end}
  )

Marking a window done is an insert rather than an update, because the timestamp is part of the sorting key and a ReplacingMergeTree would not order the versions otherwise. The comment above that insert says so, which saved the next person an afternoon.

A fifth approach exists in one repo and is worth knowing about. Instead of tracking uploads, every document is stamped with the run time and anything the run did not touch is deleted at the end:

delete_by_query(index=..., body={"query": {"range": {
    "updated_at": {"lt": int(time.time()) - (12 * 60 * 60)}}}},
    conflicts="proceed")

Idempotent writes

Every watermark scheme above leans on something it never states out loud: that sending the same document twice is harmless. Nothing here hashes document content to check. The whole thing rests on the Elasticsearch _id being the ClickHouse row key, which makes a resend an overwrite.

uploadToElastic(docs, { getId: (item) => item.id })   // the default, never overridden

That makes every retry path safe, including the one that re-sends an entire bulk batch after a partial failure. It also means the whole design rests on one field being present.

Bulk paths

A finished document is assembled from sources that do not arrive together. Identity and caption come from the nightly content pass. The counters move all day. The image embedding lands hours later from a separate GPU box. Write the whole document on each arrival and whichever writer runs last wins, silently erasing the other two. Hence three write shapes, one per kind of arrival.

The scripted upsert is the answer to re-syncing a video whose view count moved without destroying the vector attached to it:

ctx._source.views = params.views;
ctx._source.likes = params.likes;
ctx._source.comments = params.comments;
ctx._source.shares = params.shares;
ctx._source.saves = params.saves;
ctx._source.engagement_rate = params.engagement_rate;
ctx._source.updated_at = params.updated_at;

Embeddings take the opposite stance. They are written with doc_as_upsert: false, so a vector whose document has not been indexed yet is counted as not-found and skipped rather than creating a stub document with nothing but a vector in it.

Partial failures

There is a trap underneath all three of those write paths. A bulk request returns HTTP 200 even when individual documents inside it failed, and the detail sits in a per-item status array most clients never read. Treat the 200 as success and you lose documents at a rate low enough that no dashboard will ever show it.

for (const item of response.items) {
  if (item.update?.status === 200) {
    result.success++;
    if (item.update._id) result.successIds.push(item.update._id);
  } else if (item.update?.status === 404) {
    result.notFound++;                       // expected: doc not indexed yet
  } else {
    result.failed++;
    if (item.update?.error) console.error(`Update failed for ${item.update._id}:`, item.update.error);
  }
}

Triaging 404 separately from a real failure matters because the two need opposite responses. A missing parent document will exist after the next content sync. A mapping error will not fix itself.

The same bug was written twice, independently, in two repos. Both marked ids as uploaded before knowing whether the push worked.

Both are permanent silent data loss in their broken form, and both are invisible in aggregate metrics, because the loss is a fraction of a percent of documents on a run that otherwise reports success.

Retries are three attempts with linear backoff, re-sending the whole batch. That is only acceptable because of the deterministic ids in the previous section.

Mappings

Everything so far is about getting documents in. What they cost once they are in is decided by the mapping, and the default is to pay for everything: an inverted index on every field, plus a columnar copy of each so it can be sorted on, whether or not anything ever sorts on it. So every field here gets an explicit answer to two questions. Can it be searched, and can it be sorted or aggregated.

// searchable text
mentions:         { type: 'keyword', doc_values: false },
caption_keywords: { type: 'text', analyzer: 'whitespace_analyzer' },
hashtags:         { type: 'text', analyzer: 'whitespace_analyzer' },

// sortable, doc_values on
views:            { type: 'long' },
engagement_rate:  { type: 'float' },

// retrieved but never queried
video_url:        { type: 'keyword', index: false, doc_values: false },
music_title:      { type: 'keyword', index: false, doc_values: false },

One product goes further and disables indexing on the history array entirely. The 30 points ship to the browser inside _source and cost nothing in index structures:

video_counts: {
  enabled: false,
  properties: { video_count: { type: 'long' }, at: { type: 'long' } },
}

That optimisation later came close to destroying the index. It is the first entry in failure modes.

Query context

With the index shaped properly, the remaining way to make a search slow is to ask for it wrong. Elasticsearch scores the clauses you put in must and caches the ones you put in filter. In a faceted search, nine of the ten clauses want the second one.

Put a follower-count range in must and you are asking Elasticsearch to compute a relevance score for a numeric comparison, on every matching document, and then throw that score away when the results get sorted by follower count anyway.

The clean version assembles two arrays and lets the empty one fall away:

return {
  bool: {
    must:   mustClauses.length   > 0 ? mustClauses   : undefined,
    filter: filterClauses.length > 0 ? filterClauses : undefined,
  },
};

Only free text search goes into must. Every range, every terms list and every boolean toggle goes into filter.

Two details that are easy to get wrong and appear in these repos. A bool with both arrays empty serializes to {} and needs a match_all guard. And a nested clause holding nothing but exact predicates still belongs in filter context, which one product gets wrong for its entire video filter block.

Latency

All of that tuning was worth doing, and none of it was the biggest win. The query itself was never the slow part. Here is the same search deployed two ways, once behind a serverless function and once on the box running Elasticsearch:

Moving the API onto the Elasticsearch box was worth roughly 20x end to end and required no query changes at all. The service is a small Bun and Hono process whose entire job is to be close to the cluster.

Everything else measured on the serving side:

That last row is the whole argument for the split, written as a configuration value. A ten minute timeout on an interactive search is an admission that queries were taking minutes, and every cause is visible in the same file: all clauses in query context, deep from offsets, exact total counting, an 85-value terms clause from a hardcoded genre taxonomy, and full time series returned in _source.

Pagination

Pagination is where a search that feels instant on page one falls over on page fifty. Every one of these products paginates the naive way, with from and size, and not one uses search_after, a scroll or a point in time. That makes the max_result_window default of 10,000 a hard ceiling, and each product runs into it differently.

The last row is a pager hardcoded to 200 pages whatever the result count, so the UI offers the same 200 pages for a query matching three rows and one matching three million.

Bulk export is where from and size actually hurts. One export pages at 5,000 up to 100,000 results, so its final request is from: 95000, size: 5000, which makes every shard sort 100,000 documents to return the last page. Export by id list takes the sane route and batches 100 ids at a time.

No sort in any of these products has a tiebreaker field, so two documents with equal sort values can swap places between page requests. Deep pagination is non-deterministic everywhere, quietly.

Vector search

Image search arrived last and forced a decision I got wrong the first time: where the vectors live. The first attempt put them in a dedicated vector database beside Elasticsearch. The second put them in Elasticsearch itself. Having built both, the comparison is not close.

The mapping for the native version:

embedding: {
  type: 'dense_vector',
  dims: 1152,
  index: true,
  similarity: 'cosine',
  index_options: { type: 'hnsw', m: 16, ef_construction: 200 },
}

And the fusion, which is the part worth copying:

// every constraint moves into the knn filter so vector search only
// returns docs that also match the text and range criteria
const knnFilterClauses = [...filter];
if (must.length > 0) knnFilterClauses.push(...must);
if (knnFilterClauses.length > 0) searchParams.knn.filter = { bool: { must: knnFilterClauses } };

// with all constraints on the knn, a separate text search would only
// dilute the results
searchParams.query = { match_all: {} };

Note what this does to the same clauses in the two modes. They score in a keyword search. In a vector search they move into knn.filter and stop scoring altogether, and sorting is disabled so the kNN order survives to the response.

Three operational details that cost real time:

• kNN has no from. Paginating means refetching the whole prefix and slicing in the client, so page 50 of a 20-per-page image search fetches 1,000 hits and discards 980.
• The loader checks vector length on read and drops rows that do not match, because a wrong-dimension vector is a mapping error at index time rather than a bad result at query time.
• Changing embedding model means changing dimensions. Going from 768 to 1152 was a server side _reindex with a script that strips the old field: ctx._source.remove('embedding'), run with wait_for_completion: false and polled.

The model arc across these products is worth recording, because each step was a full reindex: OpenCLIP at 512 dimensions in an external store, then SigLIP2 at 768 in Elasticsearch, then SigLIP2 at 1152. The text tower of the same model encodes the query, so a typed phrase lands in the same space as the thumbnails.

Generated types

Two schemas, two systems, and a loader in the middle that has to agree with both. Keeping that agreement by hand lasts about a month. So both schemas are read from the live systems and turned into validators: ClickHouse types from system.columns, Elasticsearch types from indices.getMapping().

The generated schema then does double duty as the query planner. Removing a field from an ignore list drops it from the generated SELECT list and from the row validation in the same edit.

The generators exist, and three drift bugs survived them, which is the honest result:

A generator that is not wired into CI is documentation with extra steps. All three of these are caught by running the generator and diffing.

Failure modes

Five that cost real time. Two of them were caused by decisions described earlier in this article, which is the honest argument for writing any of it down.

The optimisation that nearly emptied the index

Disabling indexing on the history array means a range query against it matches nothing. The prune job wanted documents with no recent data point, which is the obvious must_not of that query, and a must_not of something that matches nothing matches everything.

The guard is a refusal rather than a warning: the job computes the fraction it is about to delete and aborts above 10%. First real run removed 1.17% and 1.44% of two indices, which is what a correct run looks like.

One character inverting the sync

An incremental window was written updated_at < now() - INTERVAL 1 DAY and fixed two commits later to >. In between, the sync selected exactly the rows that had not changed recently.

Forgetting FINAL

Every source table is a ReplacingMergeTree, so a read without FINAL returns unmerged duplicate versions and counts that move as background merges run. The symptom reached the UI as a catalogue total that appeared to halve.

Liveness is not progress

A scraper reported Up 19 hours while returning 401 on 328,231 requests in five minutes and writing almost nothing. A container health check passes in that state. The monitor that replaced it compares row counts per calendar day against the same hour yesterday.

The first version of that monitor then reported a critical failure over 18 rows in 30 minutes, on a day the scraper had written 1.3 million. It was punishing a scraper for finishing early.

No aliases, so no atomic swap

Index names are hardcoded strings in every repo here. Nothing writes through an alias, so there is no blue and green swap, no atomic cutover, and a partial run leaves an index half old and half new with nothing marking it as such. Changing a mapping means recreating the index in place.

This is the one piece of the design that has no defence. An alias costs nothing to add and turns a reindex into a pointer move.

Constants

The numbers this pipeline actually runs on.

The two client timeouts are that long because these are loader clients, not serving clients. A bulk of 25,000 documents against an index doing a force merge deserves ten minutes. A user-facing search does not, and the serving API sets 60 seconds.

What I would keep

Six products in, the parts of this that earned their place are shorter than the article suggests.

• Write the rule down where it gets acted on. "Elasticsearch is a published view, never the record" is a comment in a delete script. That is the one place somebody needs to believe it, at the moment they are deciding whether a destructive operation is safe.
• Derive the document id from the row key. It is one line and it makes every retry, every re-run and every crash recovery idempotent for free. The two days one syncer shipped without it duplicated an entire index on every run.
• Read the per-item statuses on every bulk response. Not the HTTP code. Track only the ids the cluster confirmed, and an unconfirmed document reappears in the next pass instead of vanishing.
• Put the API on the same box as the cluster if the numbers let you. Nothing else here came close to 20x for zero code changes.

The one I would change is aliases. Every index name in every one of these repos is a hardcoded string, so a mapping change means recreating the index in place and a half-finished run leaves the index half old and half new with nothing marking it. An alias costs an afternoon and turns that into a pointer swap. I still have not done it.

The other half of this system, the pipeline that fills ClickHouse in the first place, is written up in Tracking Half a Billion TikTok Sounds.