Kafka to ClickHouse: Streaming Ingestion, Duplicates & Tombstones Guide

Kafka topics reach ClickHouse in four ways: ClickHouse's own Kafka table engine, ClickPipes on ClickHouse Cloud, the ClickHouse Kafka Connect sink, or a managed pipeline that runs the consumers for you. ClickPipes is ClickHouse's recommended path on Cloud. The table engine suits a self-hosted server that should run nothing extra, the Connect sink suits teams that already operate Kafka Connect, and a managed pipeline fits when nobody wants to tune consumers or Kafka is one of several sources.

Two constraints shape every design. Most paths deliver at least once, so tables must absorb replayed messages, and a keyed topic with tombstones needs a ReplacingMergeTree table rather than an append-only one. ClickHouse also wants large, infrequent inserts, not one per message. Estuary covers this route with a Kafka capture that reads topics continuously into a durable collection, and a native ClickHouse materialization whose table writes follow a sync schedule.

Choosing a Kafka to ClickHouse method

Where ClickHouse runs, who should operate the consumers and how duplicates are handled decide the method.

The table engine is at-least-once; ClickPipes and the Connect sink are too, unless you turn on their exactly-once modes. Vector and custom consumer code also work, mostly for log pipelines or custom processing.

How Kafka messages become ClickHouse inserts

ClickHouse's three native methods run the same loop and differ mainly in where the consumers live: inside the ClickHouse server (table engine), in a Cloud service (ClickPipes), or on Connect workers writing over HTTP (the sink).

• A consumer group reads the topic, one consumer per partition at most, so partitions cap parallelism. A Kafka data pipeline can run many groups on one topic, each with its own offsets.
• Deserialization: JSON needs nothing else. Confluent-serialized Avro and Protobuf begin with a five-byte header, a zero byte then the schema ID, which the consumer looks up in the registry.
• Buffering collects messages into blocks, because every insert writes at least one new part for background merges to absorb.
• The insert lands a block in a MergeTree-family table.
• The offset commit follows the insert. A consumer that dies between the two leaves the block to be read again, which is why most paths are at-least-once.

Topic, partition and offset identify a message, so they are the natural duplicate check for event topics; keyed topics use the message key.

Kafka requirements

Every method is a Kafka client, so the cluster must accept one from wherever the consumers run.

Then settle four things per topic:

1. Format and registry. Avro and Protobuf need the producers' schema registry; <topic>-key and <topic>-value subjects follow Confluent's default TopicNameStrategy.
2. Keyed or not. A topic whose keys carry current state (often compacted, with null-value tombstones for deletes) needs a different table from an append-only event stream.
3. Partitions, which cap consumer parallelism for every method.
4. Retention, which must cover the longest outage you plan to recover from: consumers can only replay what the brokers still hold.

Grant the consumer READ and DESCRIBE on the topics and its consumer group.

ClickHouse requirements

ClickHouse needs a database, a user that can create and insert into tables, and a table design that matches the topic.

• Network path. The table engine connects out from the ClickHouse servers. The Connect sink writes over HTTPS on 8443 (8123 unencrypted) and refuses servers older than 23.3. ClickPipes exists only in ClickHouse Cloud, which accepts TLS connections only.
• Event topics:MergeTree, or ReplacingMergeTree with _partition and _offset last in ORDER BY, so a replayed message collapses into its original row.
• Keyed topics:ReplacingMergeTree(version, is_deleted) ordered by the message key (is_deleted needs 23.2+), queried with FINAL.
• ORDER BY up front: an existing sorting key accepts only columns created in the same ALTER.

Setting up Kafka to ClickHouse replication

This walk-through uses the Kafka table engine, available on every deployment; on ClickHouse Cloud, ClickHouse recommends ClickPipes instead.

1. Check the topic: format, registry, partitions, keyed or not.
2. Create the target table as the requirements section describes.
3. Set authentication on the Kafka table (kafka_security_protocol, kafka_sasl_mechanism, kafka_sasl_username, kafka_sasl_password) or in the server's <kafka> librdkafka configuration.
4. Create the Kafka engine table with kafka_broker_list, kafka_topic_list, kafka_group_name and kafka_format: JSONEachRow, or AvroConfluent with format_avro_schema_registry_url; Protobuf needs a schema file in kafka_schema. Keep kafka_num_consumers within the partition count and the server's physical cores.
5. Keep kafka_handle_error_mode = 'default', so a malformed message stops the block with an error instead of being skipped. To continue past bad messages, configure a dead letter queue or persist the error rows as described in troubleshooting.
6. Create the materialized viewTO the target, selecting _partition and _offset with the payload.
7. Never SELECT from the Kafka table. A read consumes what it returns. DETACH the view to pause; ATTACH resumes from committed offsets.
8. Validate. Compare row counts with each partition's offsets, look for repeated (_partition, _offset) pairs, delete a test key on a keyed topic, and watch consumer lag.

sql
CREATE TABLE events_queue
(
    event_id   String,
    user_id    UInt64,
    event_type LowCardinality(String),
    created_at DateTime64(3)
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'broker-1:9092,broker-2:9092',
         kafka_topic_list = 'events',
         kafka_group_name = 'clickhouse-events',
         kafka_format = 'JSONEachRow',
         kafka_num_consumers = 3,
         kafka_handle_error_mode = 'default';

CREATE TABLE events
(
    event_id   String,
    user_id    UInt64,
    event_type LowCardinality(String),
    created_at DateTime64(3),
    _partition UInt64,
    _offset    UInt64
)
ENGINE = ReplacingMergeTree
ORDER BY (event_type, created_at, _partition, _offset);

CREATE MATERIALIZED VIEW events_mv TO events AS
SELECT event_id, user_id, event_type, created_at, _partition, _offset
FROM events_queue;

What happens after the initial load?

• Replays. After a crash or rebalance, at-least-once methods insert a block again. An event table ordered by (…, _partition, _offset) collapses the copies under FINAL; a plain MergeTree keeps them.
• Updates on keyed topics replace older rows only in a ReplacingMergeTree table ordered by the key. With the default partitioner, a key's messages share a partition, so the offset can be the version, until someone adds partitions: Kafka then remaps keys without moving existing data.
• Tombstones. ClickPipes can delete matching rows (exactly-once mode only) or, as it recommends for large workloads, mark them with _is_deleted. The Connect sink lists deletes as unsupported, and the table engine's docs don't describe null-value messages, so test those paths.
• Compaction. A consumer far behind on a compacted topic may see only each key's latest value, and tombstones are dropped after delete.retention.ms (24 hours by default), so a longer stop can miss deletes.
• Schema changes. With the table engine, add the column to the target and recreate the view; the Connect sink's auto.evolve, off by default, adds missing columns.
• Re-reading history. ClickPipes applies its start position (beginning, latest or a timestamp) only on a consumer group's first read; a full re-read needs a new group or an offset reset, and messages past retention are gone.

Kafka to ClickHouse data type mapping

ClickHouse maps Avro types itself; JSON values are cast to whatever the target column declares.

Decimals in JSON topics

A producer that writes 19.99 as a JSON number has already turned it into a double, and toDecimal64 on a float truncates: ClickHouse's own example turns 1.15 into 1.14. Send money as strings or Avro decimal instead; converting text, as in toDecimal64(price_text, 2), is exact.

Performance, latency and ClickHouse cost

Freshness comes from how fast blocks fill and flush; cost comes from parts, merges and where the consumers run.

Insert size and part count

ClickHouse asks for at least 1,000 rows per insert, ideally 10,000 to 100,000, at about one insert per second. A quiet topic flushed every few hundred milliseconds makes many small parts until inserts fail with Too many parts. Clients that can't batch should use async_insert=1, wait_for_async_insert=1, which ClickHouse strongly recommends over fire-and-forget. ClickPipes closes a batch at 100,000 rows, a memory-based limit or 5 seconds, so a larger replica builds larger blocks.

Consumers and partitions

• Table engine: several consumers' inserts share one thread unless kafka_thread_per_consumer = 1, and each consumer adds buffers and librdkafka threads to a server that is also answering queries.
• ClickPipes and the Connect sink divide partitions among workers or tasks; ClickPipes works best when partitions roughly match its insert workers, which scale with replica size and count.
• FINAL merges duplicates in memory at query time; on large keyed tables, argMax with GROUP BY can be cheaper.

Production considerations

• Retention vs recovery. A table rebuilt by re-reading Kafka can only go back as far as retention.
• Source impact. Each pipeline is its own consumer group and reads every message, so a second destination doubles broker reads.
• Monitoring. Alert on consumer lag per partition, active parts in system.parts, error-table or system.dead_letter_queue rows, and time since the last insert.
• Security. A read-only Kafka principal, a ClickHouse user limited to one database, and TLS on both legs.

Common problems and troubleshooting

Too many parts from small Kafka batches

Inserts fail with Too many parts (…). Merges are processing significantly slower than inserts, usually on quiet topics or many consumers flushing separately. Count active parts:

sql
SELECT table, count() AS active_parts
FROM system.parts
WHERE active AND database = currentDatabase()
GROUP BY table
ORDER BY active_parts DESC;

Cheapest first: raise the block size or flush interval, consolidate consumers, or switch the writer to async inserts. Avoid a PARTITION BY that splits each block, since every partition gets its own part.

Duplicate rows after a consumer restart or rebalance

The same message appears twice. Check whether the copies share an offset:

sql
SELECT _partition, _offset, count() AS copies
FROM events
GROUP BY _partition, _offset
HAVING copies > 1
LIMIT 10;

Shared offsets mean an at-least-once replay: query with FINAL on a table ordered by (…, _partition, _offset), or enable an exactly-once mode. ClickPipes' exactly-once deduplicates by insert token only within the table's deduplication window (10,000 blocks or one hour by default), so widen it for long replays. Different offsets mean the producer sent twice; dedupe on a business key.

Ingestion stalls on one malformed message

In the default error mode the table engine throws on a message it can't parse, retries the block, and lag climbs. Use kafka_handle_error_mode = 'stream' (route _error != '' rows to their own table) or 'dead_letter_queue' (rows go to system.dead_letter_queue), or kafka_skip_broken_messages = N to skip up to N bad messages per block. The Connect sink's equivalent is errors.tolerance=all with errors.deadletterqueue.topic.name; ClickPipes keeps malformed records in a <table>_clickpipes_error table for 7 days and carries on.

Using Estuary for Kafka to ClickHouse

For this route, the Apache Kafka capture writes each topic to a collection, and the ClickHouse materialization turns each collection into a table over the native protocol, self-hosted or Cloud. Against the concepts above:

• Formats: Avro and Protobuf through a Confluent-compatible schema registry (references only for Protobuf); JSON with or without one.
• Keys: taken from the latest <topic>-key schema unless a key field is nullable, a union, a float, an array or a map. Otherwise, and on every topic read without a registry, the key is _meta/partition plus _meta/offset: one row per message.
• Tombstones: a null value is captured as a delete (_meta/op = d). It replaces the earlier row only on a registry key; on a partition-and-offset key it is one more row.
• Offsets: a new binding starts at the earliest offset on the broker. Progress lives in Estuary's checkpoints, not in Kafka, so consumer-group lag tools won't show it; the task log reports messages behind every 5 minutes.
• Security: SASL PLAIN or SCRAM, or MSK IAM with an access key, over TLS with system certificates; neither connector offers mutual TLS, a custom CA, Kerberos or an SSH tunnel. Estuary's security page states SOC 2 Type II and HIPAA compliance and shows a GDPR badge.
• Writes:ReplacingMergeTree(flow_published_at) ordered by the collection key, so query with FINAL. Rows are staged, then moved into the target by partition, a metadata-only step re-applied after a restart; on standard tables, repeats collapse by key.
• Schedule: 30 minutes by default; 0s writes as each transaction completes.
• Deletes: soft by default, so FINAL still returns rows marked d in _meta/op; Hard Delete inserts _is_deleted = 1 rows that FINAL hides.
• Types: fractional JSON numbers and Avro decimals (captured as decimal strings) become Float64; Avro timestamps become DateTime64(6, 'UTC'), clamped before 1925; nested records become JSON text in String columns.

Implementing Kafka to ClickHouse with Estuary

These steps adapt Estuary's general data flow guide to this route.

1. Complete the prerequisites in Kafka requirements and ClickHouse requirements. Estuary also needs:
   • Every advertised broker address reachable from Estuary's IP addresses; on MSK, allow them in the VPC configuration.
   • For upserts and deletes by key, a <topic>-key schema that fits the key rules.
   • ClickHouse 25.3 or later, the connector's grants (including SELECT on system.columns, system.parts and system.tables), and the native port open to Estuary.
2. Create the Kafka capture. On the Sources page, click New Capture, choose Apache Kafka, name it, and fill in the endpoint properties:
   • Bootstrap Servers: comma-separated host:port pairs.
   • Credentials: SASL (User & Password), with SASL Mechanism (choose PLAIN for Confluent Cloud; the form starts blank), Username and Password; or AWS MSK IAM, with AWS Access Key ID, AWS Secret Access Key and AWS Region.
   • TLS Settings: system certificates.
   • Schema Registry: Confluent Schema Registry (Schema Registry Endpoint, Schema Registry Username, Schema Registry Password), or No Schema Registry with Capture Messages in JSON Format Only checked.

3. Select topics. Click Next. Output Collections lists one binding per Topic.
   • Remove topics you don't need, including other brokers' internal topics.
   • Check each collection's key: _meta/partition and _meta/offset means one row per message.
   • Keep Automatically keep schemas up to date on.

Click Save and publish. Each binding reads from the earliest retained offset, then follows new messages.

4. Create the ClickHouse materialization. On the capture's details page, click Materialize, choose ClickHouse, and fill in the endpoint properties:
   • Address: host:port; without a port, 9440, or 9000 when SSL Mode is disable.
   • Database, and Authentication: Username and Password.
   • Hard Delete: on where tombstones should vanish from FINAL results; off keeps them as rows to filter on _meta/op.

5. Choose the sync schedule.Sync Frequency defaults to 30 minutes; 0s writes as each transaction completes, leaving more parts to merge. Timezone, Fast Sync Start Time and Fast Sync Stop Time limit frequent writes to working hours, with 4-hour syncs outside them.

6. Shape each table. Click Next, pick a naming convention, and review each binding under Source Collections:
   • Table: the ClickHouse table name.
   • Delta Update: for partition-and-offset event topics, a plain MergeTree table without the lookup standard bindings run before each write. Leave it off for keyed topics, or upserts and deletes stop.
   • Partition By (advanced): only on a value that never changes for a row, such as event time. On a standard table, flow_published_at splits a key's versions across months, and ClickHouse merges only within a partition.
   • For exact amounts, set castToString on the field: Avro decimals keep their text, and JSON numbers become the shortest text that round-trips (1.15 stays 1.15), ready for toDecimal64(amount, 2) in a view.
7. Publish. Click Save and publish. Estuary creates the tables and writes the backfill without waiting for the schedule.
8. Verify that data is landing.
   • The capture details page shows PRIMARY in its Shard Information card; a collection opened from its Collections list shows Data Preview.
   • The materialization details page charts Data Read.
   • In ClickHouse, run SELECT count() FROM events FINAL, then the checks in Setting up Kafka to ClickHouse replication on _meta/partition and _meta/offset.

How Estuary handles replays, Kafka retention and extra destinations

Each message is written once to a collection, JSON files in your own bucket or Estuary's trial storage, and ClickHouse tables are built from it, so most rebuilds never touch the brokers.

Rebuilding ClickHouse tables after the topic has expired

Trial storage keeps about 20 days of collection data; a bucket you own keeps all of it.

No consumers or offsets of your own

To get Kafka data into ClickHouse without running consumers yourself, let the capture run them: nothing executes on the ClickHouse servers, there's no Connect cluster, and the capture can be split across shards that each own a fixed share of partitions. Its offsets stay in its own checkpoints, so it never moves another application's consumer group. Where Kafka only carries database changes to ClickHouse, an Estuary database capture can write ClickHouse directly, without Kafka.

ClickHouse and another destination from one topic read

A second materialization on the same collections, Snowflake or Apache Iceberg for example, reads their history and then new documents, while Kafka still serves one consumer.

When another approach may make more sense

• ClickHouse Cloud, Kafka the only source: ClickPipes. It is managed, supports mutual TLS, IAM roles for MSK and the AWS Glue Schema Registry, and can delete rows on tombstones in exactly-once mode.
• Kafka Connect in production, or exactly-once on self-hosted ClickHouse: the Connect sink with exactlyOnce=true, which keeps per-partition state in KeeperMap. Tables must exist first (named after topics, or mapped with topic2TableMap), and deletes are unsupported.
• A small self-hosted pipeline: the table engine, accepting consumers on the servers' CPU, at-least-once delivery, and recovery limited to an offset reset within retention.
• Logs and observability: Vector, which parses and routes events on the way.
• ClickPipes required by policy: point a ClickPipe at Estuary's ClickHouse Kafka API connector (Dekaf), which serves collections as Kafka topics to ClickHouse Cloud.

Related resources

• Estuary: Apache Kafka capture, ClickHouse materialization, ClickHouse via Dekaf and ClickPipes, sync schedules, collection keys, backfilling data, Kafka source, ClickHouse destination
• Kafka: log compaction, producer partitioner, adding partitions, Confluent wire format, Amazon MSK ports
• ClickHouse: Kafka options, Kafka table engine, Connect sink, ClickPipes for Kafka and its best practices, ReplacingMergeTree, insert strategy, async inserts
• Vector: Kafka source, ClickHouse sink

To try the route, see the Kafka to ClickHouse integration or create a Kafka capture in the Estuary dashboard. Pricing: estuary.dev/pricing.