Me, and the firehose

James Greenhill — Member of Token Burning Staff, PostHog.

PostHog is open source. One platform, every tool:

Product analytics

Session replay

Feature flags

Data warehouse

Web analytics

Error tracking

Experiments

Surveys

Data pipelines

LLM analytics

Logs

PostHog AI

…and many more

All the analytical data lives in ClickHouse.

Team ClickHouse

trillions

of events

petabytes

of data

Kafka → CH, the classic recipe

One box, two jobs, fighting

What that looked like from the dashboard

Plan A: build an inserter service

The hard part was never the consumer loop. It was the schema, which an inserter makes you own in two places:

ClickHouse DDL

properties Map(String, String),
mat_browser String MATERIALIZED
    properties['$browser']

Go (or whatever)

Properties map[string]string
MatBrowser string // keep in
                  // sync, forever

• Every column change ships twice: DDL, then code, in the right order.
• Types drift between the two. ClickHouse feeding ClickHouse is one type system.
• Two codebases, two on-call surfaces, one schema.

A "cluster" is just a logical list

In ClickHouse, a cluster is just a <remote_servers> entry — and one config can declare many. Nodes can live anywhere, run anything, and be added at will.

<remote_servers>
    <posthog>                       <!-- storage cluster: 10×3 on EC2, NVMe -->
        <shard><replica><host>data-1</host></replica> ... </shard> ...
    </posthog>
    <ingestion-events>              <!-- a SEPARATE cluster: stateless K8s pods -->
        <shard><replica><host>chi-ingestion-events-0</host></replica></shard> ...
    </ingestion-events>
</remote_servers>

EC2 storage and stateless K8s ingestion are separate clusters — just two entries in the same list. The pods are their own cluster; their Distributed tables target posthog to forward writes. Different hardware, bridged by the Distributed engine — never one mixed cluster.

The inserter we wanted... is a ClickHouse node

Every feature we'd have built into the inserter, ClickHouse already ships. So run those three objects on a node that stores nothing.

Four objects, split across two roles

The MV reads from the Kafka table and writes to the writable Distributed table. The Distributed table forwards to the MergeTree on the data nodes. Done.

Same schema system as everything else. All four objects live in our existing ClickHouse migrations. A schema change is one migration, and each statement is applied to the role it belongs on.

The only place that actually stores rows

CREATE TABLE IF NOT EXISTS groups
(
    group_type_index UInt8,
    group_key        VARCHAR,
    created_at       DateTime64,
    team_id          Int64,
    group_properties VARCHAR,
    _timestamp       DateTime,
    _offset          UInt64
)
ENGINE = ReplicatedReplacingMergeTree(
    '/clickhouse/tables/{shard}/posthog.groups', '{replica}', _timestamp)
ORDER BY (team_id, group_type_index, group_key)

Plain replicated MergeTree. It has no idea Kafka exists. This is the only stateful piece.

A Distributed table is the "INSERT client"

CREATE TABLE IF NOT EXISTS writable_groups
(
    group_type_index UInt8, group_key VARCHAR, created_at DateTime64,
    team_id Int64, group_properties VARCHAR, _timestamp DateTime, _offset UInt64
)
ENGINE = Distributed('posthog_single_shard', 'default', 'groups');

Non-sharded target → point the Distributed table at posthog_single_shard. Sharded target → use the full posthog cluster with a sharding key:

ENGINE = Distributed('posthog', 'default', 'sharded_events', sipHash64(distinct_id));

The consumer

CREATE TABLE IF NOT EXISTS kafka_groups
(
    group_type_index UInt8, group_key VARCHAR, created_at DateTime64,
    team_id Int64, group_properties VARCHAR
)
ENGINE = Kafka(warpstream_ingestion,            -- named collection (broker list)
    kafka_topic_list = 'groups',
    kafka_group_name = 'group1',
    kafka_format     = 'JSONEachRow');

Broker lists come from named collections — which is how we run MSK and WarpStream side by side (more later). One consumer per INGESTION_SMALL node; more on the hotter tiers.

The wire between consumer and writer

CREATE MATERIALIZED VIEW IF NOT EXISTS groups_mv
TO writable_groups
AS SELECT
    group_type_index, group_key, created_at,
    team_id, group_properties,
    _timestamp, _offset
FROM kafka_groups;

That's the whole inserter. One statement wires the consumer to the writer.

The whole path, then we delete the pod

The pod holds no data. Delete it and the uncommitted rows are simply re-read from Kafka.

One migration, role-aware placement

operations = [
    run_sql_with_exceptions(GROUPS_TABLE_SQL(),          node_roles=[NodeRole.DATA]),
    run_sql_with_exceptions(GROUPS_WRITABLE_TABLE_SQL(), node_roles=[NodeRole.INGESTION_SMALL]),
    run_sql_with_exceptions(KAFKA_GROUPS_TABLE_SQL(),    node_roles=[NodeRole.INGESTION_SMALL]),
    run_sql_with_exceptions(GROUPS_TABLE_MV_SQL(),       node_roles=[NodeRole.INGESTION_SMALL]),
]

No ON CLUSTER. run_sql_with_exceptions fans each statement to exactly the hosts whose macro hostClusterRole matches — so the MergeTree lands on storage, and Kafka/MV/Distributed land on the stateless tier.

Ingestion is tiered by throughput

One NodeRole per tier. Each is its own logical cluster of stateless pods, sized to the topics it drains.

More roles, same shape: shufflehog, ai_events, aux, ops, sessions, logs, endpoints … Adding capacity means adding cheap, stateless pods.

Tuning the Kafka table engine is the fiddly part

ENGINE = Kafka(warpstream_ingestion,
    kafka_topic_list = 'events', kafka_group_name = 'clickhouse',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers       = 4,    -- ≤ topic partitions  AND  ≤ physical cores
    kafka_thread_per_consumer = 1);   -- without this, all 4 squash onto ONE thread

One flag, one thread or four

Storage vs ingestion, by the numbers

Ingestion outnumbers storage by node count — but costs a fraction, because there's no data to hold or protect.

The config that proves there's nothing to lose

hostClusterRole: ingestion
raid_volumes: []        # no NVMe / EBS data disks
backup_tables: []       # nothing to back up
clickhouse_disk_balancer_enabled: false

volumeTemplates:
  - name: ingestion
    reclaimPolicy: Delete   # PVC dies with the pod
    resources:
      requests: { storage: 10Gi }   # local cache only

Pods get recycled and lose their disk — and it doesn't matter. The data is in Kafka and on the storage cluster; the ingestion tier is pure compute.

A pod dies, one panel twitches

ClickHouse already reports on itself

-- Is every consumer alive, assigned, and polling?
SELECT hostName() AS pod, table, consumer_id,
       length(assignments.partition_id) AS partitions,
       now() - last_poll_time           AS secs_since_poll,
       num_messages_read, num_rebalance_revocations,
       exceptions.text[-1]              AS last_error
FROM clusterAllReplicas('ingestion-events', system.kafka_consumers)

-- Throughput per pod ($__timeFilter & friends: Grafana ClickHouse data source macros)
SELECT $__timeInterval(event_time) AS time, hostName() AS pod,
       sum(ProfileEvent_KafkaMessagesRead) / $__interval_s AS msgs_per_s
FROM clusterAllReplicas('ingestion-events', system.metric_log)
WHERE $__timeFilter(event_time)
GROUP BY time, pod ORDER BY time

Same shape, different sink

Every stateless node is the same core — a Kafka table engine → materialized view, no MergeTree. The only thing that changes is what the MV writes into.

Partition by the key your consumer aggregates on

Downstream is propdefs (property-defs-rs): it tracks every project + event + property key seen, buffers them, and flushes idempotent upserts into Postgres.

Same rows, same consumers, no lock queue

Swapping Kafka under a live firehose

Migrating MSK → WarpStream with zero downtime: run a fully parallel write path per backend — its own Kafka table engine → MV → Distributed table, separate consumer groups — both landing in the same data table.

Cut over by adding or removing a whole leg. No inserter to rewrite, no dual-write code — just DDL.

Sharp edges

The cost of this pattern is schema management. Every object — the data table, the Kafka table engine, the MV, the Distributed writable table — is schema we own and migrate, now across multiple node roles.

Tinybird's ingestion pain list, resolved

Source: Javi Santana, “I've operated petabyte-scale ClickHouse® clusters for 5 years”, Tinybird blog.

The payoff