Skip to main content

Kafka to ClickHouse: Topics via Apache Iceberg

Kafka to ClickHouse usually means a Kafka engine table with a materialized view per topic, ClickPipes on ClickHouse Cloud, or a Kafka Connect sink. All three load MergeTree tables that only ClickHouse reads, and updates need ReplacingMergeTree or FINAL to collapse. OLake Go reads the topics on a schedule as a change data capture (CDC) stream, keeps one row per key, and writes Apache Iceberg tables into an Apache Polaris catalog.

ClickHouse attaches that catalog as a database and queries the tables in place. Polaris is an open source Iceberg REST (Representational State Transfer) catalog you run yourself, with the files in your own bucket. DuckDB and Trino read the same tables through the same endpoint.

By Shubham Satish BaldavaUpdated October 30, 2026

Key takeaways

  • OLake Go turns Kafka topics into Iceberg tables in Polaris: each sync reads from the last committed offset and upserts messages on a deduplication key, so a repeated key updates its row instead of adding one.
  • ClickHouse attaches the whole Polaris catalog as one database with DataLakeCatalog and returns one current row per key, with no FINAL and no ReplacingMergeTree.
  • The same tables, written once by OLake Go, are also queryable from DuckDB and Trino through the same catalog. In the run on this page all three engines returned identical rows.

The short answer. To stream Kafka to ClickHouse as Iceberg, run OLake Go with the Kafka cluster as the source and an Apache Polaris catalog as the destination.

Put each topic in CDC Only mode with Upsert on a deduplication key. The first sync reads every retained message; every later sync reads only newer messages. ClickHouse, DuckDB and Trino query the same tables.

How Kafka messages reach ClickHouse here​

There is no separate snapshot phase, because the topic is the history. The first sync runs with a new consumer group, which has no committed offset, so it reads every message the broker still holds: on this page all 48.

OLake Go commits the Iceberg snapshot first and the consumer-group offsets second. A sync that fails mid-way therefore re-reads its messages on the next run, which Upsert mode absorbs.

Every later sync resumes from the committed offsets and reads only what arrived since. A message whose deduplication key already exists retires the old row with a positional delete and writes the new one. Keys seen for the first time become new rows, and ClickHouse applies the deletes while reading.

One thing to keep an eye on: retention is the replay window. Kafka deletes messages past retention.ms, and OLake Go cannot read what is gone. A job paused longer than the retention of its topics silently misses the expired messages.

What you need​

  • A Kafka cluster on version 2.4 or later that OLake Go can reach, with JSON messages (or Avro plus a schema registry) on the topics you want. Amazon Managed Streaming for Apache Kafka (MSK), Confluent Cloud, Redpanda and self-hosted Kafka all work; the run on this page used a single-node Apache Kafka 4.3.1 in KRaft mode in Docker.
  • A field that identifies each entity in every message, such as id, when you want one row per key. Without it, run the topic in Append mode.
  • An Apache Polaris server and a bucket, both reachable from OLake Go and from ClickHouse. The run on this page used Polaris 1.8.0 and MinIO in Docker.
  • ClickHouse 26.1 or later, where the DataLakeCatalog engine reads REST catalogs. The run on this page used the 26.8 LTS release in Docker.
  • OLake UI from the Docker Compose install, on OLake UI v0.7.1 or later with the OLake Go Kafka source on v0.12.1 or later, which is when Upsert mode for Kafka arrived. The run on this page used OLake Go v1.1.0.
  • Docker, and about 40 minutes.

Step 1: Prepare the Kafka topics​

Most readers point OLake Go at topics they already have. If you want to follow along with the exact data from this page, produce messages shaped like these first:

The two sample topics used on this page

Both topics have one partition and carry JSON values keyed by the entity id. The scripts ship with Apache Kafka under bin/.

Sample topics and messages
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --partitions 1
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 1

# key<TAB>value, one message per line
printf '1\t{"id":1,"name":"Aarav Mehta","email":"aarav@example.com","city":"Mumbai","tier":"gold","created_at":"2026-10-29T15:22:04+00:00","updated_at":"2026-10-29T15:22:04+00:00"}\n' \
  | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic customers --property parse.key=true --property key.separator=$'\t'

printf '1\t{"id":1,"customer_id":2,"product":"Headphones","amount":99.19,"status":"shipped","created_at":"2026-10-29T15:22:04+00:00","updated_at":"2026-10-29T15:22:04+00:00"}\n' \
  | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders --property parse.key=true --property key.separator=$'\t'

# Then produce whatever you like. This page started with 8 customers and 40 orders.

Kafka is already a change stream, so there is no log to switch on. OLake Go reads each topic from its retained offsets and commits the position it reached in a consumer group. Four things must be in place before the first sync.

RequirementValueWhy it matters
Message formatJSON values, or Avro with a schema registryOLake Go turns each message into a row. Anything else (Protobuf, plain text) cannot be mapped to columns.
A stable field per entityFor example id in the payload, or the message keyUpsert mode needs a deduplication key to know which earlier row a message replaces. Keep it in every message and never change it.
A consumer group for OLake Go onlyOne group per jobOLake Go stores its offsets there. Sharing the group with another consumer, or between two jobs, moves offsets under its feet.
Retention longer than your longest pauseFor example 7 daysKafka deletes messages past the retention limit. A job paused longer than that cannot read them, and they are gone.

If the cluster enforces access control lists (ACLs), the user OLake Go connects with needs Read and Describe on the topics and Read on its consumer group. This is what the setup looked like for the run on this page, with the scripts that ship with Apache Kafka; Redpanda's rpk topic create and rpk security acl create do the same:

On the Kafka cluster
# 1. The topics, with retention long enough to cover a pause
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --partitions 1 --config retention.ms=604800000
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 1 --config retention.ms=604800000

# 2. If ACLs are on: a principal for OLake Go (one command per resource type)
kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:olake \
--operation Read --operation Describe --topic customers --topic orders
kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:olake \
--operation Read --group olake_<job_name>

Each message needs the entity's identifier in its value, and the key is the natural place to repeat it. A record for the sample customers topic looks like this, with _kafka_key set to the same id:

One message on the customers topic (key: 3)
{"id": 3, "name": "Liam Carter", "email": "liam@example.com", "city": "Austin", "tier": "gold",
 "created_at": "2026-10-29T15:22:04+00:00", "updated_at": "2026-10-29T15:22:04+00:00"}
Retention is your replay window

A topic with retention.ms of one day keeps one day of messages. If the job is paused for two, the first day is gone before OLake Go reads it, and nothing in the destination tells you. Set retention from the longest pause you can tolerate, and alert when a job has not run for a good fraction of it.

Step 2: Set up the Polaris catalog​

ClickHouse reads Iceberg tables through a catalog. Apache Polaris is the one this page uses: an open source REST catalog that you run next to your bucket.

OLake Go writes into it, and ClickHouse, DuckDB and Trino read from it with the credentials Polaris hands out. The Polaris catalog page lists every engine that reads it.

CatalogOLake Go writesClickHouse readsVerified on this page
Apache PolarisYesYes: DataLakeCatalog with catalog_type restYes, end to end

Apache Polaris is an open source Iceberg REST catalog. It keeps the table metadata pointers, hands out short-lived storage credentials to engines, and applies its own role-based access control. OLake Go writes to it through the standard Iceberg REST API, and ClickHouse, Trino, DuckDB and Spark read from the same endpoint. The files live in an S3 bucket or an S3-compatible store that you own.

Three things need to exist before OLake Go can write: a Polaris server, a catalog inside it that points at your bucket, and a principal (a client id and secret) with the right to manage that catalog. The run on this page used Polaris 1.8.0 and MinIO in Docker; on AWS, swap the MinIO endpoint for a plain S3 bucket and an IAM role.

docker-compose.yml (Polaris and MinIO, for a local test)
services:
  minio:
    image: minio/minio
    hostname: lake-minio
    command: server /data --console-address ":9001"
    environment: { MINIO_ROOT_USER: minioadmin, MINIO_ROOT_PASSWORD: minioadmin }
    ports: ["9000:9000", "9001:9001"]
  polaris:
    image: apache/polaris:latest
    depends_on: [minio]
    ports: ["8181:8181", "8182:8182"]
    environment:
      AWS_REGION: us-east-1
      AWS_ACCESS_KEY_ID: minioadmin
      AWS_SECRET_ACCESS_KEY: minioadmin
      POLARIS_BOOTSTRAP_CREDENTIALS: POLARIS,root,<root-secret>
      polaris.realm-context.realms: POLARIS

Polaris stores its own state in memory with this setup, so it starts empty every time; a production deployment points it at Postgres. Once it is up, create the bucket, the catalog and the principal through the management API. root is the bootstrap principal; the olake principal is the one OLake Go and the engines use.

Polaris, once
P=http://localhost:8181

# 1. A bucket for the tables (MinIO; on AWS create it in the console or with aws s3 mb)
aws --endpoint-url http://localhost:9000 s3 mb s3://lakehouse

# 2. A root token
TOKEN=$(curl -s $P/api/catalog/v1/oauth/tokens -H 'Polaris-Realm: POLARIS' \
  -d grant_type=client_credentials -d client_id=root -d client_secret=<root-secret> \
  -d scope=PRINCIPAL_ROLE:ALL | jq -r .access_token)
H=(-H "Authorization: Bearer $TOKEN" -H 'Polaris-Realm: POLARIS' -H 'Content-Type: application/json')

# 3. The catalog: where the files go, and how Polaris and the engines reach the store
curl -s "${H[@]}" $P/api/management/v1/catalogs -d '{"catalog": {
  "name": "olake", "type": "INTERNAL", "readOnly": false,
  "properties": {"default-base-location": "s3://lakehouse/olake"},
  "storageConfigInfo": {"storageType": "S3", "allowedLocations": ["s3://lakehouse/olake"],
    "endpoint": "http://<minio-host>:9000/", "endpointInternal": "http://lake-minio:9000",
    "pathStyleAccess": true, "region": "us-east-1"}}}'

# 4. A principal for OLake Go and the engines. The response holds clientId and clientSecret: keep them.
curl -s "${H[@]}" $P/api/management/v1/principals -d '{"principal": {"name": "olake"}}'

# 5. Roles and the grant that lets the principal create namespaces and tables in the catalog
curl -s "${H[@]}" $P/api/management/v1/principal-roles -d '{"principalRole": {"name": "olake_role"}}'
curl -s "${H[@]}" $P/api/management/v1/catalogs/olake/catalog-roles -d '{"catalogRole": {"name": "olake_catalog_role"}}'
curl -s -X PUT "${H[@]}" $P/api/management/v1/principals/olake/principal-roles -d '{"principalRole": {"name": "olake_role"}}'
curl -s -X PUT "${H[@]}" $P/api/management/v1/principal-roles/olake_role/catalog-roles/olake -d '{"catalogRole": {"name": "olake_catalog_role"}}'
curl -s -X PUT "${H[@]}" $P/api/management/v1/catalogs/olake/catalog-roles/olake_catalog_role/grants -d '{"type": "catalog", "privilege": "CATALOG_MANAGE_CONTENT"}'

What each block is for:

  • endpoint is the store address Polaris hands to engines along with vended credentials; endpointInternal is the address Polaris itself uses. On AWS S3 leave both out. On MinIO both are required, or Polaris tries to reach AWS and fails. Keep the trailing slash on endpoint: ClickHouse 26.8 and 26.10 build the object path from it, and without the slash every read fails with a 404 (the fix is explained under troubleshooting on this page).
  • Polaris vends credentials through the store's STS endpoint. MinIO exposes one, so nothing else is needed. For a store without STS, add "stsUnavailable": true and give Polaris static keys to hand out.
  • CATALOG_MANAGE_CONTENT lets the principal create namespaces and tables and read and write their data. Scope it to a namespace instead when several teams share the catalog.
  • The principal's secret is shown once, in the response to step 4. It is what goes into OLake Go as <client_id>:<client_secret>.
Hostnames with underscores break Polaris

Polaris is a Java service, and Java's URI parser rejects hostnames that contain an underscore. A MinIO container named lab_minio makes every table creation fail with host must not be null or Expected scheme-specific part at index 5: http:. Use a hyphen in the hostname (lake-minio), as above.

You can check the principal outside OLake Go by exchanging its credentials for a token, the same call OLake Go makes on every sync:

Token check
curl -s http://localhost:8181/api/catalog/v1/oauth/tokens -H 'Polaris-Realm: POLARIS' \
  -d grant_type=client_credentials -d client_id=<client_id> -d client_secret=<client_secret> \
  -d scope=PRINCIPAL_ROLE:ALL

A good principal returns a JSON object with an access_token. Running Polaris on Kubernetes, with Postgres behind it, or with an external identity provider is covered on the Polaris catalog page.

Step 3: Connect Kafka to OLake Go​

  1. In the OLake UI, open Sources and click Create Source. Pick Kafka as the connector, give the source a name and keep the OLake Go version on the latest release.
  2. The Enable CDC toggle under Change data capture (CDC) reads not supported for Kafka and stays off. That is expected: the topic is the change stream, and the job's CDC Only sync mode reads it.
  3. Under Connection details, fill in Bootstrap Servers (host:port, comma-separated for several brokers), a Consumer Group ID for this job only, and a Topic Pattern regular expression that matches only the topics you want. Pick the Security Protocol the cluster uses: PLAINTEXT, SASL_PLAINTEXT or SASL_SSL, with the SASL mechanism and credentials when needed. Turn on Use Schema Registry only for Avro topics.
  4. Leave Additional configuration at its defaults, or set Max Threads to the number of partitions you want read in parallel.
  5. Click Set up source. OLake Go pings the brokers. Topics are matched against the pattern at discovery, when the job is created; if the pattern matches nothing, the Streams page comes up empty and the sync log names every topic it skipped.

Here is the source form from the run on this page. Kafka ran in another container, so the bootstrap server is host.docker.internal:19092; the topic pattern ^(customers|orders)$ keeps discovery to the two sample topics.

OLake UI Edit source form for Kafka on OLake Go v1.1.0 with bootstrap server host.docker.internal:19092, a consumer group, topic pattern ^(customers|orders)$ and security protocol PLAINTEXT
The Kafka source in OLake UI: bootstrap server, a consumer group for the job, and a topic pattern that matches the two sample topics.

Set up source pings the brokers before anything is saved. The consumer group is created on the first sync; it does not need to exist yet.

The full field reference, including SASL and schema-registry options, is on the Kafka connector page.

Step 4: Configure the Iceberg destination​

  1. Open Destinations and click Create Destination.
  2. Pick Apache Iceberg as the connector and Polaris as the catalog type.
  3. Choose OAuth2 as the authentication type. REST Catalog URI is the Polaris catalog endpoint (http://<polaris-host>:8181/api/catalog), Polaris Catalog Name the catalog from the setup (olake), REST Auth URI the token endpoint, and REST Credential the principal's <client_id>:<client_secret>.
  4. Fill in the store: AWS Region, and for MinIO the S3 Endpoint with the AWS Access Key and AWS Secret Key. On AWS S3 leave the endpoint empty.
  5. Click Create. OLake Go exchanges the credentials for a token, creates a namespace and a test table through Polaris, writes to the bucket and drops the test table again.

Here is the destination form from the run on this page. Polaris and MinIO ran in other containers, so both addresses use host.docker.internal; the credential is the olake principal from step 2.

OLake UI Edit destination form for Apache Iceberg with catalog type Polaris, OAuth2, REST catalog URI on port 8181, catalog name olake and the MinIO endpoint
The Polaris destination in OLake UI: OAuth2 with the principal from step 2, and the MinIO endpoint for the files.

Polaris Configuration Parameters

ParameterSample ValueDescription
REST Catalog URI
required		http://<REST_ENDPOINT>:8181/api/catalogEndpoint URL for the Polaris REST catalog service.
Authentication Type
required		oauth2Authentication method OLake Go uses for Polaris REST catalog requests. Set to match your catalog's authentication configuration.
Polaris Catalog Name
required		<POLARIS_CATALOG_NAME>Name of the Polaris catalog to use, OLake Go passes this to the Polaris REST catalog as the warehouse identifier, so catalog operations are routed to that catalog and its pre-configured S3 storage and IAM settings.
REST Auth URI
required		http://<REST_ENDPOINT>:8181/api/catalog/v1/oauth/tokensURL of the REST catalog’s OAuth2 token endpoint. OLake Go uses this with REST Credential to request an access token during OAuth2 client-credentials authentication.
REST Credential
required		<olake_user_client_id>:<olake_user_client_secret>Specifies the client ID and secret for OAuth2, formatted as client_id:client_secret. Used with REST Auth URI when OLake Go requests an access token from the catalog’s auth service. Not needed when using REST Token or other auth types.
AWS Region
required		ap-south-1AWS Region where your S3 bucket exists. Required so OLake Go calls the correct regional S3 endpoint.
REST ScopePRINCIPAL_ROLE:ALLRequired for OAuth2 client-credentials auth. Scopes are space-separated. Used with REST Auth URI and REST Credential for OAuth2 client-credentials auth. Values are catalog-specific. Not needed when using REST Token or other auth types.
S3 Endpointhttp://<S3_ENDPOINT>:9000S3 API endpoint for writing Iceberg data files. Leave empty for AWS S3 but required for S3-compatible storage like MinIO.
AWS Access KeyXXXAWS access key ID for authentication. Optional if using IAM roles or instance profiles.

If using IAM:
Docker Compose: add the required IAM environment variables under x-envs in your compose file. See Service Environment Variables for setup details.
Kubernetes: set up pod IAM in Cloud Integration.
AWS Secret KeyXXXAWS secret access key for authentication. Optional if using IAM roles or instance profiles.

If using IAM:
Docker Compose: add the required IAM environment variables under x-envs in your compose file. See Service Environment Variables for setup details.
Kubernetes: set up pod IAM in Cloud Integration.
Enable Arrow WritesfalseWrites data and delete files using Apache Arrow based writer and registers them in Iceberg.
Catalog Nameolake_icebergName of the Iceberg catalog OLake Go registers tables under. Defaults to olake_iceberg if left empty.
Catalog Name Supported for v0.3.5 and above
For the catalog name, OLake Go only supports lowercase letters and underscores. Spaces and special characters are not supported.

Step 5: Create the job and run the first sync​

The target query engines on this job are ClickHouse, DuckDB and Trino. Kafka streams offer one sync mode, CDC Only, and the upsert type on this job is Positional Deletes, which all three engines read. What you decide on the Streams page is the ingestion mode and the deduplication key.

From OLake Go v1.1.1 and OLake UI v0.8.2 Kafka upserts can also write deletion vectors, which ClickHouse 26.8 does not read.

  1. Open Jobs and click Create Job. Give it a name and a schedule. Every Day at a fixed hour, or a cron expression, both work; the schedule is how often the change stream is drained, not how often a full copy runs.
  2. Select the source and the destination you created.
  3. Under Advanced Settings, set Target Query Engines to ClickHouse, plus DuckDB and Trino if they will read the tables too. OLake Go then resolves an Iceberg delete format every one of those engines can read, so you never pick one by hand.
  4. Click Next. OLake Go discovers the topics and opens the Streams page.
  5. Tick the topics to replicate. For each one, set the sync mode to CDC Only and the ingestion mode to Upsert. Under Dedup Keys, pick id: that is the field OLake Go uses to match a later message to the row it replaces. Leave the Upsert Type as OLake Go resolved it (Positional Deletes in the run on this page).
  6. Click Create Job, then open the job's menu and choose Sync Now for the first run.
OLake UI Streams Selection page with the topics namespace, customers and orders selected, CDC Only, Upsert, dedup key id and Positional Deletes on Iceberg V2
Each topic: CDC Only, Upsert on the id field, and Positional Deletes on Iceberg v2. The Iceberg DB name at the top is editable.
OLake UI Job Logs and History for kafka_to_clickhouse showing two completed syncs, 25 seconds and then 23 seconds
Job history: the 25 s first read of both topics, then the 23 s sync of the new messages.

On this page the first run read 8 customer and 40 order messages (48 records, 9.00 KB) and finished in 25 seconds, container start included. OLake Go names the Iceberg namespace after the job and the source namespace. The namespace for every Kafka topic is topics, so kafka_to_clickhouse produced kafka_to_clickhouse_topics.

Step 6: Query the tables in ClickHouse​

There is no registration step. ClickHouse attaches the whole Polaris catalog as one database with the DataLakeCatalog engine, lists every table OLake Go committed, and reads the files with the credentials Polaris vends. One thing to know before the first query: ClickHouse has one level of namespace, so the Iceberg namespace and table name go together inside backticks.

clickhouse-client
SET allow_database_iceberg = 1;

CREATE DATABASE polaris
ENGINE = DataLakeCatalog('http://<polaris-host>:8181/api/catalog')
SETTINGS
catalog_type = 'rest',
warehouse = 'olake',
catalog_credential = '<client_id>:<client_secret>',
oauth_server_uri = 'http://<polaris-host>:8181/api/catalog/v1/oauth/tokens',
auth_scope = 'PRINCIPAL_ROLE:ALL',
vended_credentials = true,
region = 'us-east-1';

SHOW TABLES FROM polaris;

-- What OLake Go created (namespace = <job name>_<source database>_<source schema>)
SELECT id, name, city, tier
FROM polaris.`kafka_to_clickhouse_topics.customers`
ORDER BY id;

vended_credentials = true makes ClickHouse read the files with the short-lived keys Polaris hands out per table, so no store keys sit in ClickHouse. To use static keys instead, replace it with storage_endpoint, aws_access_key_id and aws_secret_access_key. The database is read-only: INSERT, ALTER and RENAME DATABASE return NOT_IMPLEMENTED.

OLake Go adds four columns to every table: _olake_id (a hash of the primary key), _op_type, _olake_timestamp and _cdc_timestamp. _op_type is r for a row from the first snapshot, c for an insert, u for an update and d for a delete. Filter them out of dashboards, or use them to see when a row last changed.

clickhouse-client querying polaris.kafka_to_clickhouse_topics.customers through the DataLakeCatalog engine and returning 10 rows, one per key
ClickHouse 26.8 reading the customers table through Polaris after the second sync: 10 rows, one per key, with no FINAL.

Step 7: Produce new messages and sync again​

To see the change path work, the run on this page produced 12 more messages after the first sync. Seven carry new keys; five repeat an existing key with a changed value.

Messages produced between the two syncs (key, then value)
# customers: two new keys, two repeated keys
9   {"id":9,"name":"Maya Patel","email":"maya@example.com","city":"Pune","tier":"silver", ...}
10  {"id":10,"name":"Lucas Silva","email":"lucas@example.com","city":"São Paulo","tier":"bronze", ...}
3   {"id":3,"name":"Liam Carter","email":"liam@example.com","city":"Austin","tier":"gold", ...}
8   {"id":8,"name":"Zara Ali","email":"zara@example.com","city":"Abu Dhabi","tier":"silver", ...}

# orders: five new keys (41 to 45), three repeated keys (1, 4, 7) with status "delivered"
41  {"id":41,"customer_id":9,"product":"Laptop","amount":1299.0,"status":"placed", ...}
1   {"id":1,"customer_id":2,"product":"Headphones","amount":99.19,"status":"delivered", ...}

The second sync read 12 records (2.26 KB) from the two topics and finished in 23 seconds end to end. For orders that meant one new data file with 8 rows: the 5 new keys plus the new versions of the 3 repeated keys.

One small positional delete file retires the 3 old versions. OLake Go's commit report in the sync log shows it directly: addedDataFiles=1, addedRecords=8, addedPositionalDeleteFiles=1, addedPositionalDeletes=3.

OLake UI sync log filtered to the Iceberg commit report for orders, showing addedDataFiles 1, addedPositionalDeleteFiles 1, addedRecords 8 and addedPositionalDeletes 3
The second sync in the OLake UI log: one data file and one positional delete file for orders.

customers has the same shape with 4 rows and 2 deletes. ClickHouse shows the result on the next query, with no refresh and still one row per key.

Liam Carter is gold, Zara Ali moved to Abu Dhabi, and the two new customers are there. The _op_type column shows how each row arrived (c for a first-seen key, u for a repeated key) and _kafka_offset shows which message it came from.

Schema changes. When a new field appears in the messages, OLake Go adds it as a column on the next sync and commits the new schema to Polaris. ClickHouse shows it the next time the table is opened.

A field whose type changes in an incompatible way fails the sync rather than corrupt the table. Full rules are on the schema evolution page.

Query the same tables from DuckDB and Trino​

This is the point of writing Iceberg instead of loading ClickHouse tables. Nothing is exported, copied or refreshed: both engines read the files OLake Go wrote, through the same Polaris endpoint.

DuckDB attaches Polaris as an Iceberg REST catalog with the principal's client id and secret; DuckDB fetches the token itself. On a laptop, where the store address Polaris hands out may not resolve, add an S3 secret for the store and turn credential vending off:

DuckDB 1.5.5
INSTALL iceberg; LOAD iceberg;
INSTALL httpfs;  LOAD httpfs;

CREATE SECRET polaris_secret (
TYPE iceberg,
CLIENT_ID '<client_id>', CLIENT_SECRET '<client_secret>',
OAUTH2_SERVER_URI 'http://<polaris-host>:8181/api/catalog/v1/oauth/tokens',
OAUTH2_SCOPE 'PRINCIPAL_ROLE:ALL'
);
-- MinIO, or any store whose address differs from the one Polaris vends
CREATE SECRET store_secret (
TYPE s3, KEY_ID '<access key>', SECRET '<secret key>',
ENDPOINT '<minio-host>:9000', URL_STYLE 'path', USE_SSL false, REGION 'us-east-1'
);

ATTACH 'olake' AS polaris (
TYPE iceberg,
ENDPOINT 'http://<polaris-host>:8181/api/catalog',
SECRET polaris_secret,
ACCESS_DELEGATION_MODE 'none'
);

SHOW ALL TABLES;
SELECT id, name, city, tier FROM polaris.kafka_to_clickhouse_topics.customers ORDER BY id;

With Polaris on AWS S3, drop the S3 secret and the ACCESS_DELEGATION_MODE line: DuckDB then reads with the credentials Polaris vends. DuckDB reads equality and positional deletes, and deletion vectors from duckdb-iceberg 1.5.3 onward.

DuckDB 1.5.5 returned the same 10 customers and 45 orders as ClickHouse in the run on this page.

An orders roll-up matched the latest message per key in all three engines: 45 orders for 25579.19 in total. By status it gave 16 delivered orders for 8659.68, 18 placed for 10522.26, 11 shipped for 6397.25.

Data types: Kafka JSON to Iceberg to ClickHouse​

A Kafka message has no schema of its own, so OLake Go infers one from a sample of messages at discover time and maps it to Iceberg types. ClickHouse then maps Iceberg types to its own. Four Kafka columns are added to every row:

In the JSON valueIceberg (written by OLake Go)ClickHouse showsNotes
Integer ("id": 3)longNullable(Int64)
Decimal ("amount": 99.19)doubleNullable(Float64)The orders.amount sums on this page matched the messages to the cent in every engine; round at query time when you need exact decimal arithmetic.
ISO 8601 timestamp string ("2026-10-29T15:22:04+00:00")timestamptz (UTC)Nullable(DateTime64(6, 'UTC'))Strings that parse as timestamps are stored as timestamps.
Other stringsstringNullable(String)
_kafka_keystringNullable(String)The message key, base64-encoded (3 arrives as Mw==).
_kafka_partition, _kafka_offsetint, longNullable(Int32), Nullable(Int64)Where the row's latest message sat on the topic.
_kafka_timestamptimestamptzNullable(DateTime64(6, 'UTC'))Produce time or broker append time, depending on the topic's message.timestamp.type.

Types on the left come from the Kafka connector reference. The ClickHouse side comes from the ClickHouse Iceberg table engine page (checked 2026-10-30).

Cost and maintenance​

  • Storage is your bucket: you pay S3 or MinIO rates for the Parquet and delete files, and nothing to Polaris or ClickHouse for holding them.
  • Compute. OLake Go runs wherever you run Docker, for the length of a sync. ClickHouse spends CPU only when you query; a DataLakeCatalog query reads Parquet over the network with no MergeTree index, so load a hot slice into a MergeTree table when a dashboard needs sub-second answers.
  • Maintenance. A topic synced every minute adds a small file and a delete file per run, so plan compaction. OLake Fusion does this for the catalogs OLake Go writes to, and Spark's rewrite_data_files works through the same Polaris endpoint.
  • Retention disk. Seven days of a busy topic is real disk on the brokers. Size for it, or shorten retention once the job runs reliably every hour.

Troubleshooting​

1. The source test cannot reach the brokers​

failed to ping kafka brokers: unable to dial: dial tcp 192.168.65.254:19999: connect: connection refused

The address in Bootstrap Servers is not reachable from inside the OLake Go container. For a broker on the same machine use host.docker.internal, not localhost. Make sure the broker also advertises a listener the container can resolve; Kafka's advertised.listeners decides that.

2. The Streams page is empty​

Skipping topic customers (does not match topic_pattern: ^nothing_here$)

The topic pattern is a full regular expression anchored by you. ^orders$ matches one topic, ^orders matches every topic that starts with it, and a pattern with a typo matches nothing. Fix the pattern on the source and run discovery again.

3. ClickHouse lists the tables but every query returns a 404​

Code: 499. DB::Exception: Failed to get object info for key olake/.../metadata/00003-....metadata.json in bucket lakehouse: No response body.. HTTP response code: 404

The object key is doubled. ClickHouse 26.8 and 26.10 join the table location onto the storage endpoint, then fail to strip it again when the endpoint has no trailing slash. Set the Polaris catalog's endpoint (and storage_endpoint in ClickHouse, if you use static keys) to a value ending in /, for example http://minio:9000/.

4. The destination test fails at the token exchange​

OLake Go's destination test exchanges the principal's credentials at /api/catalog/v1/oauth/tokens before anything else. Three causes cover nearly every failure: a wrong client id or secret, a principal with no role that carries CATALOG_MANAGE_CONTENT on the catalog, or a wrong Polaris Catalog Name. Run the token check from the catalog setup; it returns the same error Polaris gives OLake Go.

5. A sync ends early after a consumer-group rebalance​

A rebalance in OLake Go's consumer group, usually because another process joined it, makes the sync stop cleanly with what it has read. Nothing is lost: the messages read so far are committed, and the next sync resumes from the committed offsets. Give each job its own consumer group and never point another consumer at it.

How this compares with other ways to move Kafka data to ClickHouse
MethodFreshnessUpdates and deletesWhere the data landsWhat you run
ClickHouse Kafka table engine and a materialized viewSecondsThrough ReplacingMergeTree and FINAL, or your own mergeNative ClickHouse tables; ClickHouse is the only readerA Kafka engine table and a materialized view per topic
ClickPipes for KafkaSecondsAppends; collapse with a table engineNative ClickHouse Cloud tablesNothing on your side; ClickHouse Cloud only
Kafka Connect ClickHouse sinkSeconds to minutesAppends; collapse with a table engineNative ClickHouse tablesA Kafka Connect cluster and the sink connector
Managed ELT (Fivetran, Estuary, Airbyte Cloud)Minutes, by planVaries by toolClickHouse tables, a second copy you pay to move and storeNothing; priced per row, per GB or per connector
OLake Go to Iceberg on Polaris (this guide)On a schedule down to one minuteYes, upsert on a key with Iceberg row-level deletesApache Iceberg tables that ClickHouse, DuckDB, Trino and Spark all readOne Docker Compose stack, open source; your bucket plus your own compute

The Kafka table engine is the simpler choice when you need seconds of latency into MergeTree and ClickHouse is the only reader. Writing Iceberg once, with one row per key, avoids paying to move and store a second copy. That matters as soon as another engine needs the same rows, or you want to own the files and the catalog.

When not to use this​

  • You need seconds of latency. OLake Go drains the topic on a schedule, one minute at the shortest. Use the Kafka table engine or ClickPipes for a continuous feed.
  • Every query must hit MergeTree speed. A DataLakeCatalog query reads Parquet with no primary index. Keep the Iceberg tables as the source and load the hot slice into a MergeTree table on a schedule.
  • Messages are not JSON or Avro. Protobuf, plain text and binary payloads cannot be mapped to columns.
  • The partition count will change while the job runs in Upsert mode. The same key would then hash to a different partition, and concurrent partition readers could leave a stale row behind. Finish the repartition, then create the job.

FAQs​

How do I get Kafka topics into ClickHouse without the Kafka table engine or ClickPipes?

Run OLake Go (open source, Apache 2.0) with the Kafka cluster as the source and an Apache Polaris catalog as the Iceberg destination. Each sync reads the topics from the last committed offset, upserts the messages on a deduplication key, and commits Iceberg tables that ClickHouse reads through its DataLakeCatalog engine. No Kafka engine table, no materialized view and no ClickHouse Cloud account sit in the path.

Does OLake Go keep one row per key, or one row per message?

One row per key when the stream runs in Upsert mode with a deduplication key, which is what this page uses: a later message with the same key replaces the earlier row. ClickHouse then needs no FINAL, no ReplacingMergeTree and no argMax to get the current row. In Append mode every message becomes a row instead.

Does ClickHouse read Iceberg tables with row-level deletes?

Yes. ClickHouse reads positional deletes, and equality deletes from 25.8. The run on this page upserted with positional deletes, and ClickHouse 26.8 returned one row per key after the repeated messages. Deletion vectors (Iceberg v3) are not read by the 26.8 LTS release.

Where does the first sync start reading a Kafka topic?

A new consumer group has no committed offset, so the first sync reads every message the broker still retains in every partition of every selected topic. The run on this page read all 48 messages on its first sync. Each later sync resumes from the offsets OLake Go committed in its consumer group, so retention is the replay window.

Why Apache Polaris rather than a ClickHouse table?

A Polaris catalog is one open endpoint that ClickHouse, DuckDB, Trino and Spark all read, with its own access control and vended storage credentials. Loading into a ClickHouse MergeTree table is faster to query but makes ClickHouse the only reader, and a second engine means a second copy. Keep the Iceberg tables as the shared source and load hot slices into MergeTree when a dashboard needs it.

Which message formats does OLake Go read from Kafka?

JSON values, and Avro values when a schema registry is configured on the source. Field types are inferred from a sample of messages at discover time: integers show up in ClickHouse as Int64, decimals as Float64, ISO 8601 timestamp strings as DateTime64 and other strings as String. Protobuf and plain-text messages are not supported.

Kafka to SnowflakeSame source, read from Snowflake.Kafka to BigQuerySame source, read from BigQuery.Kafka to Amazon RedshiftSame source, read from Amazon Redshift.Kafka to DatabricksSame source, read from Databricks.Kafka to Apache IcebergSame source, read from Apache Iceberg.Postgres to ClickHouseSame engine, replicated from Postgres.MySQL to ClickHouseSame engine, replicated from MySQL.MongoDB to ClickHouseSame engine, replicated from MongoDB.SQL Server to ClickHouseSame engine, replicated from SQL Server.DB2 to ClickHouseSame engine, replicated from DB2.Kafka connector referenceEvery field, sync mode and troubleshooting note for the source.Query engine compatibilityWhich engines read which catalogs and delete formats.All integrationsEvery source and engine pair with a tested guide.

Deeper reading on this site: