Skip to main content

Kafka to Snowflake: Topics as Apache Iceberg

Kafka to Snowflake usually means the Snowflake Kafka connector, a Kafka Connect cluster, and tables that only Snowflake can read. This guide takes a different route.

OLake Go reads the topics itself as a change data capture (CDC) stream and keeps one row per key. It writes the result once, as Apache Iceberg tables that Snowflake and every other engine can query.

With Snowflake's own Horizon catalog, the default on this page, the files sit on Snowflake-managed storage. Nothing else is in the path: no Kafka Connect, no Snowpipe, no staging table. With AWS Glue, the files sit in your S3 bucket and Snowflake reads them through a catalog integration.

By Shubham Satish BaldavaUpdated October 29, 2026

Key takeaways

  • OLake Go turns Kafka topics into Snowflake Iceberg tables: 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.
  • Through the Horizon catalog the tables land as Snowflake-managed Iceberg tables. You query them in Snowsight right away, with no CREATE ICEBERG TABLE, no external volume and no refresh.
  • The same tables, written once by OLake Go, are also queryable from DuckDB and Spark through the same catalog. In the run on this page all three engines returned identical rows.

The short answer. To stream Kafka to Snowflake as Iceberg, run OLake Go with the Kafka cluster as the source and the Snowflake Horizon 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 the messages after the committed offsets. Snowflake, DuckDB and Spark then query the same Iceberg tables, with no second copy.

How Kafka messages reach Snowflake 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. In Upsert mode the re-read is harmless; in Append mode it duplicates those rows.

Every later sync resumes from the committed offsets and reads only what arrived since. In Upsert mode a message whose deduplication key already exists in the table retires the old row with a positional delete and writes the new one.

A key seen for the first time becomes a new row. In Append mode every message is a row, which suits event logs that are never corrected.

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. Set retention from your longest tolerable pause, not from disk size.

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.
  • A Snowflake account where you can run the one-time setup as ACCOUNTADMIN. Snowflake-managed storage means no cloud bucket or IAM role on your side.
  • 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 30 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/; with Redpanda, rpk topic create and rpk topic produce -f '%k\t%v\n' do the same.

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 catalog in Snowflake​

Snowflake reads Iceberg tables from two kinds of catalog. Horizon is Snowflake's own: OLake Go writes straight into it, the tables are Snowflake-managed, and they appear in Snowsight the moment a sync commits.

AWS Glue is a catalog outside Snowflake, so the files stay in your S3 bucket. Snowflake reaches them through a catalog integration: a read-only link to that external catalog, refreshed after each sync.

Horizon is the default on this page. Glue is the right pick when Athena, Redshift or Spark on Amazon Elastic MapReduce (EMR) will read the same tables. It also keeps the files in your own bucket.

CatalogOLake Go writesSnowflake readsVerified on this page
AWS GlueYesYes: catalog integration (read-only)Documented, not yet run by us
Snowflake HorizonYesYes: native: tables are Snowflake-managed Iceberg tablesYes, end to end

Snowflake Horizon is the catalog built into every Snowflake account. OLake Go talks to it through the Iceberg REST endpoint at https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog, and the tables it creates are Snowflake-managed Iceberg tables. Writing through this endpoint needs OLake Go 0.11.0 or later.

Run the following once in a Snowsight worksheet as ACCOUNTADMIN. It creates a database for the tables, a role OLake Go writes as, a service user that holds the token, and the authentication policy the token needs.

Snowflake, once per account
USE ROLE ACCOUNTADMIN;

-- A small warehouse for your own queries, and the database OLake Go writes into.
CREATE WAREHOUSE IF NOT EXISTS OLAKE_WH WAREHOUSE_SIZE = XSMALL AUTO_SUSPEND = 60 AUTO_RESUME = TRUE;
CREATE DATABASE IF NOT EXISTS OLAKE_DEMO;
ALTER DATABASE OLAKE_DEMO SET CATALOG = 'SNOWFLAKE';
ALTER DATABASE OLAKE_DEMO SET EXTERNAL_VOLUME = 'SNOWFLAKE_MANAGED';

-- The role OLake Go uses. It needs to create schemas; tables are created inside them.
CREATE ROLE IF NOT EXISTS OLAKE_ROLE;
GRANT USAGE, CREATE SCHEMA ON DATABASE OLAKE_DEMO TO ROLE OLAKE_ROLE;
GRANT USAGE ON WAREHOUSE OLAKE_WH TO ROLE OLAKE_ROLE;
GRANT ROLE OLAKE_ROLE TO ROLE SYSADMIN;

-- A service user for OLake Go. It holds the programmatic access token.
CREATE USER IF NOT EXISTS OLAKE_SVC TYPE = SERVICE DEFAULT_ROLE = OLAKE_ROLE;
GRANT ROLE OLAKE_ROLE TO USER OLAKE_SVC;

-- Snowflake issues a token to a service user only under a network policy, unless an
-- authentication policy waives that. This one waives it. Any schema can hold the policy.
USE SCHEMA OLAKE_DEMO.PUBLIC;
CREATE AUTHENTICATION POLICY IF NOT EXISTS olake_pat_policy
  PAT_POLICY = (NETWORK_POLICY_EVALUATION = ENFORCED_NOT_REQUIRED);
ALTER USER OLAKE_SVC SET AUTHENTICATION POLICY olake_pat_policy;

-- The two values OLake Go asks for.
SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME() AS account_identifier,
       CURRENT_REGION() AS region;

-- The token. Snowflake shows the secret once; copy it straight into OLake Go.
ALTER USER OLAKE_SVC ADD PROGRAMMATIC ACCESS TOKEN OLAKE_PAT
  ROLE_RESTRICTION = 'OLAKE_ROLE' DAYS_TO_EXPIRY = 30;

What each block is for:

  • CATALOG = 'SNOWFLAKE' and EXTERNAL_VOLUME = 'SNOWFLAKE_MANAGED' on the database make every table OLake Go creates a Snowflake-managed Iceberg table on Snowflake's own storage. No S3 bucket, IAM role or external volume on your side. To keep the files in your own bucket, create an external volume and name it here instead.
  • The authentication policy is the step people miss. Without it, the ADD PROGRAMMATIC ACCESS TOKEN line fails with Network Policy is required when creating a programmatic access token for user OLAKE_SVC. In production, attach a network policy that allows OLake Go's egress IP instead of waiving the check.
  • ROLE_RESTRICTION pins the token to OLAKE_ROLE, so a leaked token cannot act as ACCOUNTADMIN. Tokens expire after DAYS_TO_EXPIRY (15 by default, 365 at most). Rotate it before then or the sync fails with an authentication error.

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

Token check
curl -s -X POST "https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog/v1/oauth/tokens" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  --data-urlencode grant_type=client_credentials \
  --data-urlencode scope=session:role:OLAKE_ROLE \
  --data-urlencode client_secret=<your PAT>

A good token returns a JSON object with an access_token. The other authentication methods (key-pair JWT, External OAuth, workload identity federation) are on the Horizon 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 the Enable CDC toggle marked not supported, bootstrap server host.docker.internal:19092, consumer group olake_kafka_to_snowflake, topic pattern ^(customers|orders)$, security protocol PLAINTEXT, schema registry off, max threads 2
The Kafka source in OLake UI: bootstrap server, a consumer group for this job only, 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 Horizon as the catalog type.
  3. Choose Programmatic Access Token (PAT) as the authentication type and fill in the fields below.
  4. Click Create. OLake Go calls the Horizon endpoint, exchanges the token and saves the destination.
OLake UI destination form for the Snowflake Horizon catalog with the PAT authentication fields

Horizon Configuration Parameters

ParameterSample ValueDescription
Authentication Type
required		Programmatic Access Token (PAT)Authentication method OLake Go uses for Horizon Catalog REST API requests. Set to match your workspace's authentication configuration.
Horizon Iceberg REST Catalog URI
required		https://<account_identifier>.snowflakecomputing.com/polaris/api/catalogSnowflake Horizon Catalog REST API endpoint.
Snowflake Database
required		SNOWThe name of the database in Snowflake.
Session Role Scope
required		session:role:<HORIZON_ROLE>Informs Snowflake which role to use for this Horizon Catalog REST session, which controls privileges for catalog and table operations. Space-separated if providing more than one.
Programmatic Access Token
required		ver:1-hint:...Snowflake programmatic access token (PAT).
Cloud Storage Region
required		AWS_AP_SOUTH_1Snowflake's account region used to route requests to the correct storage endpoint. Run SELECT CURRENT_REGION(); to know the region
Catalog NamehorizonName of the Iceberg catalog OLake Go registers tables under. Defaults to olake_iceberg if left empty.
Enable Arrow WritesfalseWrites data and delete files using Apache Arrow based writer and registers them in Iceberg.
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​

  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 Snowflake, plus DuckDB and Apache Spark 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.

Three target query engines were set on this job. Kafka streams offer one sync mode, CDC Only, so the two decisions on the Streams page are the ingestion mode and the deduplication key. OLake Go works out the Iceberg delete format all three engines can read and sets it on every stream.

OLake UI Streams Selection page with the topics namespace, customers and orders selected, sync mode CDC Only, ingestion mode Upsert, dedup key id and upsert type Positional Deletes on Iceberg format version V2
Each topic: CDC Only, Upsert on the id field, and the upsert type OLake Go resolved (Positional Deletes in this run). The Iceberg DB name at the top is editable.
OLake UI Job Logs and History for kafka_to_snowflake showing two completed syncs, the first in 44 seconds and the second in 24 seconds
Job history: the 44 s first read of both topics, then the 24 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 44 seconds, container start included. OLake Go names the destination schema after the job and the source namespace.

The namespace for every Kafka topic is topics, so kafka_to_snowflake produced kafka_to_snowflake_topics. The Iceberg DB field at the top of the Streams page changes it before the first sync.

Step 6: Query the tables in Snowflake​

There is no registration step. The schema and tables OLake Go committed through Horizon are already Snowflake-managed Iceberg tables in the database you configured. One thing to know before the first query: OLake Go creates them with quoted, lowercase names, so Snowflake needs the double quotes.

Snowsight
-- What OLake Go created (schema = <job name>_<source database>[_<source schema>])
SHOW ICEBERG TABLES IN SCHEMA OLAKE_DEMO."kafka_to_snowflake_topics";

-- Quoted identifiers: the names are lowercase, as they were in the source
SELECT "id", "name", "city", "tier"
FROM OLAKE_DEMO."kafka_to_snowflake_topics"."customers"
ORDER BY "id";

SHOW ICEBERG TABLES reports catalog_name = SNOWFLAKE, iceberg_table_type = MANAGED and external_volume_name = SNOWFLAKE_MANAGED: Snowflake treats them as its own managed Iceberg tables, not as external tables that need a refresh.

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.

Snowsight SQL file running SHOW ICEBERG TABLES on the kafka_to_snowflake_topics schema and listing customers and orders with owner OLAKE_ROLE, external volume SNOWFLAKE_MANAGED, catalog SNOWFLAKE and table type MANAGED
SHOW ICEBERG TABLES in Snowsight: both topics are now Snowflake-managed tables, created by OLAKE_ROLE through Horizon.
Snowsight query on OLAKE_DEMO.kafka_to_snowflake_topics.customers returning 10 rows with id, name, city and tier
The customers table in Snowsight after the second sync: 10 rows, one per key, on an X-Small warehouse.

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 24 seconds end to end. For customers that meant one new data file with 4 rows: the 2 new keys plus the new versions of the 2 repeated keys.

One small delete file retires the 2 old versions. OLake Go's commit report in the sync log shows it directly: addedDataFiles=1, addedRecords=4, addedPositionalDeleteFiles=1, addedPositionalDeletes=2, and orders the same shape with 8 rows and 3 deletes.

OLake UI sync log filtered to the Iceberg commit report for customers, showing addedDataFiles 1, addedPositionalDeleteFiles 1, addedRecords 4, totalRecords 12 and addedPositionalDeletes 2
The second sync in the OLake UI log: one data file and one positional delete file for customers, zero equality deletes.

Snowflake shows the result with no refresh. Liam Carter is gold, Zara Ali moved to Abu Dhabi, and the two new customers are there, with still one row per key. The _op_type column shows how each row arrived and _kafka_offset shows which message it came from:

Snowsight query on customers after the second sync showing 10 rows with _op_type c for first-seen keys, u for Liam Carter and Zara Ali, and _kafka_offset 10 and 11 for the two replaced rows
Snowsight after the second sync: the two repeated keys (u) carry offsets 10 and 11, the rest are first-seen keys (c).

Schema changes. When a new field appears in the messages, OLake Go adds it as a column on the next sync, and Snowflake shows it. Widening changes (an integer field that starts carrying decimals) are applied.

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 Spark​

This is the point of writing Iceberg instead of loading Snowflake tables. Nothing is exported, copied or refreshed: both engines read the files OLake Go wrote, through the same Horizon endpoint, with credentials vended by Snowflake.

DuckDB attaches Horizon as an Iceberg REST catalog. It wants a bearer token rather than the raw PAT, so exchange the PAT first (the curl call from the catalog setup), then attach:

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

CREATE SECRET horizon_secret (TYPE iceberg, TOKEN '<access_token from /v1/oauth/tokens>');

ATTACH 'OLAKE_DEMO' AS horizon (
TYPE iceberg,
ENDPOINT 'https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog',
SECRET horizon_secret,
ACCESS_DELEGATION_MODE 'vended_credentials'
);

SHOW ALL TABLES;
SELECT id, name, city, tier FROM horizon.kafka_to_snowflake_topics.customers ORDER BY id;

The session token lives about an hour; re-run the exchange and CREATE OR REPLACE SECRET when a query fails with an authorization error. DuckDB reads deletion vectors from duckdb-iceberg 1.5.3 onward, and only attaches REST catalogs backed by S3, S3 Tables or Google Cloud Storage, so pick an AWS- or GCP-hosted Snowflake account for this path.

DuckDB 1.5.5 shell querying horizon.kafka_to_snowflake_topics.customers and returning the same 10 rows as Snowflake
DuckDB 1.5.5 reading the Snowflake-managed table through Horizon: the same 10 rows.

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

Data types: Kafka JSON to Iceberg to Snowflake​

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.

Snowflake then maps Iceberg types to its own. Four Kafka columns are added to every row. This is what the run on this page produced:

In the JSON valueIceberg (written by OLake Go)Snowflake showsNotes
Integer ("id": 3)longNUMBER
Decimal ("amount": 99.19)doubleFLOATThe 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)TIMESTAMP_LTZ(6)Strings that parse as timestamps are stored as timestamps; set the session zone before comparing.
Other stringsstringVARCHAR
_kafka_keystringVARCHARThe message key, base64-encoded (3 arrives as Mw==).
_kafka_partition, _kafka_offsetlongNUMBERWhere the row's latest message sat on the topic.
_kafka_timestamptimestamptzTIMESTAMP_LTZ(6)The message timestamp: 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 Snowflake side from Snowflake's Iceberg data types page (checked 2026-10-14). With Normalization on, which is the default, top-level fields become columns.

Cost and maintenance​

  • Storage is Snowflake's with SNOWFLAKE_MANAGED: you pay Snowflake storage rates for the Parquet and delete files. Point the database at your own external volume to keep the files in your bucket instead.
  • Compute. OLake Go runs wherever you run Docker; it does not use a Snowflake warehouse to write. Snowflake compute is spent only when you query.
  • Maintenance. Snowflake expires old snapshots and compacts manifests on Snowflake-managed tables at no cost. Automatic data-file compaction is bundled only when Snowflake alone writes the table (Snowflake-managed Iceberg table maintenance). A topic synced every minute adds a small file and a delete file per run, so plan compaction from the writer side; OLake Fusion does this for the catalogs OLake Go writes to, and Spark's rewrite_data_files works through the same Horizon 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.
  • Token rotation. The programmatic access token expires (30 days in the setup above, 365 at most). Put the renewal on a calendar; an expired token fails the next sync at the catalog handshake, before any message is read.

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 (and Redpanda's --advertise-kafka-addr) decide that. Credentials are not checked until the ping succeeds.

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; the sync log names every topic it skipped and why.

3. Snowflake cannot find the table​

SQL compilation error:
Object 'OLAKE_DEMO.KAFKA_TO_SNOWFLAKE_TOPICS.CUSTOMERS' does not exist or not authorized.

Snowflake upper-cased an unquoted identifier. OLake Go creates the schema and tables with their lowercase topic names, so quote them: OLAKE_DEMO."kafka_to_snowflake_topics"."customers".

4. 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. If a group's offsets are damaged, create the job again with a new group id. The group name cannot be edited on an existing source in the UI.

5. The destination test fails at the catalog handshake​

OLake Go's destination test exchanges the token at /v1/oauth/tokens before anything else. Three causes cover nearly every failure: a service user without a network or authentication policy, a scope that does not match the token's ROLE_RESTRICTION, or an expired token. Run the curl exchange from the catalog setup; it returns the same error Snowflake gives OLake Go.

How this compares with other ways to move Kafka data to Snowflake
MethodFreshnessUpdates and deletesWhere the data landsWhat you run
Snowflake connector for Kafka with Snowpipe StreamingSecondsLoads only; Snowflake's overview says updates and deletes to messages might not be reflectedNative Snowflake tables, as RECORD_CONTENT and RECORD_METADATA VARIANT columns unless schematized; Snowflake is the only readerA Kafka Connect cluster with the connector, plus Snowpipe credits
Snowflake Openflow connector for KafkaSecondsNot stated in the connector docsNative Snowflake tablesAn Openflow deployment, billed in Snowflake credits
Apache Iceberg Kafka Connect sinkMinutes, by commit intervalAppend, or row-level changes when id columns and the Debezium or DMS transform are configuredIceberg tables in a catalog you runA Kafka Connect cluster with the sink, plus a catalog
Managed ELT (Fivetran, Estuary, Airbyte Cloud)Minutes, by planVaries by toolSnowflake tables, a second copy you pay to move and storeNothing; priced per row, per GB or per connector
OLake Go to Iceberg on Horizon (this guide)On a schedule down to one minuteYes, upsert on a key with Iceberg row-level deletesApache Iceberg tables that Snowflake, DuckDB, Spark and Trino all readOne Docker Compose stack, open source; Snowflake storage plus your own compute

The Snowflake connector is the simpler choice when you need seconds of latency and Snowflake 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.

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 Snowflake connector with Snowpipe Streaming for a continuous feed.
  • Messages are not JSON or Avro. Protobuf, plain text and binary payloads cannot be mapped to columns.
  • Deletes must propagate and the key is in the payload. Tombstones delete a row only when _kafka_key is the single deduplication key; on a topic whose dedup key is a payload field, a tombstone fails the sync. Key the messages by the entity id and use the key as the deduplication key, or run the topic in Append mode and filter deletes downstream.
  • 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 Snowflake without the Snowflake Kafka connector?

Run OLake Go (open source, Apache 2.0) with the Kafka cluster as the source and Snowflake's Horizon catalog as the Iceberg destination. Each sync reads the topics from the last committed offset, upserts the messages on a deduplication key, and commits Snowflake-managed Iceberg tables. No Kafka Connect cluster, no Snowpipe and no staging tables 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. In Append mode every message becomes a row, with the Kafka partition, offset and timestamp kept in columns, which suits event logs that are never updated.

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: anything the broker has already expired cannot be read.

Can OLake Go delete a Snowflake row from a Kafka tombstone?

Yes, when the deduplication key is the message key (_kafka_key) alone. A message with that key and a null value deletes the matching row. When the deduplication key comes from the payload, as on this page, a tombstone carries no payload to match, and the connector reference says a message with every dedup field missing fails the sync. Filter tombstones upstream, or key messages by the entity id and use _kafka_key as the single dedup key.

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 become NUMBER in Snowflake, decimals FLOAT, ISO 8601 timestamp strings TIMESTAMP_LTZ and other strings VARCHAR. Protobuf and plain-text messages are not supported.

Does Kafka to Snowflake with OLake Go work on Amazon MSK, Confluent Cloud or Redpanda?

Yes. OLake Go speaks the Kafka protocol with PLAINTEXT, SASL_PLAINTEXT or SASL_SSL and the PLAIN or SCRAM-SHA-512 mechanisms, which covers MSK, Confluent Cloud and self-hosted clusters. The run on this page used Apache Kafka 4.3.1 in KRaft mode. The connector was tested most on Amazon MSK.

Can I use AWS Glue instead of the Horizon catalog?

Yes. With Glue as the catalog the files stay in your S3 bucket, Athena, Redshift and EMR read them natively, and Snowflake reads them through a catalog integration as read-only tables you refresh after each sync. Horizon is the better default when Snowflake is the main reader, because the tables are Snowflake-managed and need no refresh.

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.Kafka to ClickHouseSame source, read from ClickHouse.Postgres to SnowflakeSame engine, replicated from Postgres.MySQL to SnowflakeSame engine, replicated from MySQL.MongoDB to SnowflakeSame engine, replicated from MongoDB.Oracle to SnowflakeSame engine, replicated from Oracle.SQL Server to SnowflakeSame engine, replicated from SQL Server.DB2 to SnowflakeSame engine, replicated from DB2.MariaDB to SnowflakeSame engine, replicated from MariaDB.S3 to SnowflakeSame engine, replicated from S3.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: