Me, and the firehose

Kafka → CH, the classic recipe

Kafka table engine is a consumer. A materialized view fires on each inserted block and writes into the real MergeTree table. Elegant — pull-based, resilient to spikes, no external writer.

One box, two jobs, fighting

Plan A: build an inserter service

The hard part was never the consumer loop — it was the schema. An inserter has to denormalize events into our ever-changing dynamic event schema, which we'd then maintain in two surface areas: once in Go, once in ClickHouse SQL. Wrangling prop Maps and materialized columns in one place was already painful.

And the types would drift subtly between Go and ClickHouse — versus ClickHouse feeding native ClickHouse, one type system end to end.

How far Plan A got, in full (~/inserter):

module github.com/posthog/inserter
go 1.20
// ...line 3 was never written.   (Apr 2023)

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

You probably don't need an inserter service.

ClickHouse is already a perfectly good stream processor. Give it a node with no disk to worry about.

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.

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            -- ← writes into the Distributed table
AS SELECT
    group_type_index, group_key, created_at, team_id, group_properties,
    _timestamp, _offset
FROM kafka_groups;            -- ← reads from the Kafka table engine

Selecting from the Kafka table advances consumer offsets. The MV fires per block, transforms, and pushes into writable_groups → which forwards to the data nodes. That's the whole inserter.

The whole path

Native-protocol INSERT over the cluster. The pod holds no data — kill it, replace it, scale it, nothing is lost.

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

class NodeRole(StrEnum):
    DATA              = "data"        # the sharded MergeTree storage cluster
    INGESTION_SMALL   = "small"       # 1 Kafka consumer  — low-volume topics
    INGESTION_MEDIUM  = "medium"      # 4 Kafka consumers — mid-volume
    INGESTION_EVENTS  = "events"      # the main event firehose, many shards
    SHUFFLEHOG        = "shufflehog"  # re-partitions events so each key has one consumer
    # separate clusters: AI_EVENTS, AUX, OPS, SESSIONS, LOGS, ENDPOINTS ...

Each role is its own logical cluster of stateless nodes, sized to the topic it drains. Scaling ingestion now means adding cheap, stateless pods — never touching the storage tier.

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

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.

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.

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.

The payoff

Reach for ClickHouse's own primitives before you build a service around it.

Kafka engine + materialized view + Distributed table, on a node that stores nothing, is a stream processor.