realtime-olap-clickhouse

Sub-second analytics over a live event stream — ClickHouse incremental materialised views keep every rollup fresh at insert time, so there is no scheduled batch job between "event happened" and "dashboard updated".

The problem

Product, growth and ops teams want live answers — "what's revenue this minute?", "which products are converting right now?", "where is the funnel leaking?" — over a firehose of clickstream events. The usual stack answers them with a scheduled batch job (dbt/Airflow every 5–15 min) or by scanning raw events per request. The first is stale; the second gets slow and expensive the moment the raw table hits millions of rows. You end up choosing between fresh and fast.

What this is

A serving layer that is both. Events land in a ClickHouse MergeTree table; an incremental materialised view fires on every insert, aggregating just the new block into pre-rolled tables (AggregatingMergeTree). A small FastAPI serves those rollups. Queries never touch the raw stream, so they stay in the single-digit milliseconds while the raw table grows unbounded — and the rollups are updated synchronously as data arrives, not on a cron.

The USP: fresh and fast, with no orchestration tier. The freshness comes from ClickHouse's insert-triggered MVs, not a scheduler you have to run, monitor and backfill.

Measured locally on the embedded engine (300k events, 50 runs/query):

Metric	Value
Ingest throughput	~184,000 rows/s
revenue/timeline p95	3.3 ms
products/top p95	5.0 ms
funnel p95	2.3 ms

Architecture

flowchart LR
    GEN[Event generator<br/>view · add_to_cart · purchase] -->|batched INSERT| GATE{idempotent?<br/>batch fingerprint}
    GATE -.replay.-> SKIP[skipped<br/>no double-count]
    GATE -->|new batch| E[events<br/>MergeTree · TTL prune]
    subgraph CH [ClickHouse]
        E -->|insert trigger| MV1[[revenue_mv]]
        E -->|insert trigger| MV3[[revenue_hour_mv]]
        E -->|insert trigger| MV2[[product_mv]]
        E -->|insert trigger| MV4[[category_mv]]
        E -->|insert trigger| MV5[[traffic_mv · uniqState]]
        E -->|insert trigger| MV6[[order_value_mv · quantilesState]]
        MV1 --> R1[(revenue_per_minute)]
        MV3 --> R3[(revenue_per_hour)]
        MV2 --> R2[(product_stats)]
        MV4 --> R4[(category_stats)]
        MV5 --> R5[(traffic_per_minute)]
        MV6 --> R6[(order_value_per_minute)]
        BF[[olap-backfill<br/>replays the MV's stored SELECT]] -.rebuild.-> R4
    end
    E -.invalid rows.-> DLQ[(dead_letters<br/>reject_reason · TTL)]
    R1 --> API{{FastAPI}}
    R2 --> API
    R3 --> API
    R4 --> API
    R5 --> API
    R6 --> API
    DLQ --> API
    API --> C[/health · /metrics · /stats · /funnel · /dlq<br/>/revenue/timeline · /revenue/compare · /revenue/distribution<br/>/products/top · /categories · /traffic/timeline/]
    R1 --> CLI[/olap-query CLI<br/>same rollups, JSON to stdout/]
    R4 --> CLI
    R5 --> CLI

Loading

Run it

uv sync --extra dev

# stream events into ClickHouse — materialised views update on insert
uv run olap-ingest --events 100000

# quantitative benchmark + fail-loud latency gate (exits non-zero if p95 > 1s)
uv run olap-bench --events 200000

# serve the aggregation API on http://127.0.0.1:8000
uv run olap-serve
curl -s 'localhost:8000/products/top?metric=revenue&limit=5'
curl -s 'localhost:8000/categories'                    # per-category funnel + revenue
curl -s 'localhost:8000/traffic/timeline?minutes=60'   # unique sessions/users per minute
curl -s 'localhost:8000/revenue/compare?minutes=60'    # this hour vs the previous hour
curl -s 'localhost:8000/revenue/distribution?minutes=60' # avg / p50 / p90 / p95 order value
curl -s 'localhost:8000/dlq'                           # rejected rows by reason

# or query the same rollups straight from a shell, no server (JSON to stdout)
uv run olap-query --query compare --minutes 60
uv run olap-query --query categories

# rebuild a rollup that was added after data landed (replays the MV's own SELECT)
uv run olap-backfill --view category_mv     # or --all

# tests (real embedded ClickHouse) + lint
uv run pytest
uv run ruff check . && uv run ruff format --check .

By default everything runs on embedded ClickHouse (chDB) — no server, no Docker, no cloud account. See RUNBOOK.md for the full command reference.

Serving from a real ClickHouse server

The same code drives a real server; it's one environment variable:

docker compose up -d                 # ClickHouse on :8123
export OLAP_CLICKHOUSE_HOST=localhost
uv sync --extra dev --extra server   # adds clickhouse-connect
uv run olap-ingest --events 500000
uv run olap-serve                     # /health now reports "engine":"server"

Retention, idempotency & observability

# retain only 7 days of raw events (rollups are kept regardless)
OLAP_EVENTS_TTL_DAYS=7 uv run olap-ingest --events 100000

# replaying an identical batch is a no-op — the rollups are not double-counted.
# --as-of pins the timestamps so --seed fully reproduces the batch.
export OLAP_DATA_DIR=./data
uv run olap-ingest --events 100000 --seed 42 --as-of 2026-07-07T12:00:00   # applied
uv run olap-ingest --events 100000 --seed 42 --as-of 2026-07-07T12:00:00   # skipped

# coarse hour buckets for a wide window; scrape Prometheus metrics
curl -s 'localhost:8000/revenue/timeline?minutes=1440&resolution=hour'
curl -s 'localhost:8000/metrics'
curl -s 'localhost:8000/health'   # {"status":"ok","rollup_lag_seconds":...}

Why this stack

Choice	Rationale
ClickHouse	Columnar OLAP built for exactly this: billions of rows, sub-second GROUP BY, and insert-triggered materialised views for incremental pre-aggregation.
AggregatingMergeTree + *State/*Merge	Stores partial aggregation states so counts/sums from many inserts and background merges combine correctly — the rollup is always a faithful aggregate of the raw stream.
chDB (embedded) + server, one interface	The real ClickHouse engine as a library means laptop and CI run identical SQL with zero setup; the server path is one env var away for concurrent production serving.
FastAPI	Thin, typed HTTP over the rollups; the query layer is a handful of *Merge SELECTs.
uv + ruff	Fast, reproducible packaging and one-tool lint+format.

Operational characteristics

• Fresh by construction — MVs fire on insert; no scheduler, no backfill window.
• Sub-second serving — queries read pre-aggregated rollups, never the raw table.
• Self-pruning raw stream — a MergeTree TTL drops events past OLAP_EVENTS_TTL_DAYS (default 30); the rollups have no TTL, so served numbers outlive the raw rows.
• Idempotent ingest — a batch-fingerprint ledger makes replays (retried job, at-least-once source) a no-op instead of double-counting the rollups.
• Multi-resolution — per-minute and per-hour rollups, each fed by its own MV; /revenue/timeline?resolution=hour serves wide windows without minute-row bloat.
• Slice-by-dimension — a per-category rollup (/categories) serves the merchandising cut (funnel + revenue per category) from a handful of rows.
• Unique-visitor counting — a uniqState/HyperLogLog traffic rollup (/traffic/timeline) counts distinct sessions and users per minute without retaining raw ids, combining correctly across every insert.
• Period-over-period KPI — /revenue/compare returns this window vs the prior equal window in one -MergeIf pass, reporting null when there is no baseline.
• Order-value distribution — /revenue/distribution serves avg/p50/p90/p95 basket values from mergeable quantile sketches (quantilesState), null when the window holds no purchases.
• Backfillable rollups — olap-backfill rebuilds any rollup from the raw events by replaying its MV's own stored SELECT, so a rollup added to a live system does not serve empty history.
• Terminal access — olap-query runs any serving query as JSON from a shell, over the same rollups and engine selection as the API.
• Prometheus metrics + liveness — /metrics exposition and a /health that reports rollup freshness (rollup_lag_seconds) and flips to degraded past a threshold.
• Fail-loud latency gate — olap-bench reddens CI if any query's p95 breaches budget.
• Persistent dead-letter queue — events with unknown types, missing products or negative revenue land verbatim in a dead_letters table with a reject_reason, never silently ingested and never merely dropped; triage via /dlq.
• Structured observability — every stage emits a JSON event line (event, rows_in/out, rejected, rows_per_s); see observability.py.
• Portable — identical SQL on embedded chDB or a ClickHouse server via one env var.

⚠️ Honest disclaimer

Claim	Reality
Runs on a real engine	Yes. chDB is ClickHouse compiled as a library — tests and the happy path run real MergeTree + materialised-view semantics, no stubs. The docker-compose path was also run end-to-end against a real ClickHouse 24.8 server (ingest + all API endpoints verified).
Event data	The clickstream is a synthetic generator (illustrative catalogue and view→cart→purchase probabilities). The pipeline, SQL, rollups and API are real; point olap-ingest at your own producer to use real events. The MV rollups faithfully reproduce the generator's configured conversion rates (e.g. 35.0% view→cart), which is how the correctness tests are anchored.
Latency numbers	Measured single-node on a laptop-scale dataset (300k events). They demonstrate the pre-aggregation win, not a benchmarked cluster SLA.
Scale & ops	Single-node, no auth/RBAC, no multi-shard/replica topology. The chDB serving path serialises requests behind a lock (single session); real concurrent serving uses the ClickHouse-server path.

License

MIT — see LICENSE.