Kafka → Iceberg

Kafka to Iceberg: Stream, Transform and Sink with SQL

Five ways to get Kafka topics into Apache Iceberg tables, a complete SQL walkthrough with RisingWave, and the design decisions that decide whether the table is still fast in six months: commit interval, small files, upsert vs append-only, catalog and compaction.

Last reviewed 2026-11-08. Every statement below is current RisingWave syntax; links go to the reference docs.

Try RisingWave Free Iceberg Sink Docs

SQL

Transform In Flight

Filter, deduplicate, join and aggregate Kafka data before it lands in Iceberg, with standard SQL.

Exactly-once

Iceberg Commits

Snapshots are committed per checkpoint and never replayed twice after recovery.

60s

Default Commit Interval

Tunable down to seconds. Fewer commits mean fewer small files.

Any catalog

Glue, REST, Hive, S3 Tables

Or let RisingWave host the catalog and manage the table itself.

Options

What are the ways to get Kafka data into Iceberg?

Kafka holds events for days; Iceberg holds them for years and lets every query engine read them. Moving data between the two is now a solved problem in five different ways, and the right one depends on a single question: does the data need to change shape on the way? If the answer is no, pick the simplest copier. If yes, you need a stream processor, and the rest of this page is about doing that in SQL.

Option	Transforms	Latency to Iceberg	What you operate	Pick it when
Kafka Connect Iceberg sink	Single-message transforms only. No joins, no aggregation, no dedup across messages.	Commit interval, typically minutes.	A Connect cluster plus connector config; exactly-once via a control topic.	Faithful copy of a topic, schema already final.
Confluent Tableflow / Redpanda Iceberg Topics	None. The topic is materialized as a table as-is (Tableflow can apply a registered schema).	Minutes.	Managed by the broker vendor; tied to that platform.	You are already on that platform and only need raw tables.
Apache Flink (DataStream or Flink SQL)	Full stream processing, including custom code.	Seconds to minutes, set by the checkpoint interval.	Flink cluster, state backend, checkpoints, JVM tuning, job restarts.	You need custom operators or already run Flink.
Spark Structured Streaming	Full, in micro-batches.	Minutes.	Spark cluster and scheduler; same stack as batch Spark.	Latency in minutes is fine and the team is Spark-native.
RisingWave	Full SQL: joins (including with CDC tables and Iceberg tables), aggregations, windows, dedup, UDFs.	About a second inside RisingWave; Iceberg visibility set by commit interval (60s default, tunable to seconds).	One system; state on object storage; built-in compaction and snapshot expiration.	The data has to change shape, or several tables come from one topic.

Sources: Iceberg Kafka Connect docs, Confluent Tableflow, Redpanda Iceberg Topics, Iceberg Flink docs, Iceberg Spark Structured Streaming docs, RisingWave Iceberg docs.

Walkthrough

How do you build a Kafka to Iceberg pipeline with SQL?

The example takes a clickstream topic in Avro, removes bot traffic and duplicate events, enriches each click with page metadata from PostgreSQL, and lands two tables in the lake: the clean events and a one-minute rollup. It is four steps and about sixty lines of SQL. Replace the broker, catalog and credential placeholders with your own.

Step 1: Declare the Kafka source and the reference table

A source is a continuously read stream; it does not store the topic inside RisingWave. The CDC table does store the Postgres rows and keeps them current, so the join in the next step sees page metadata changes within seconds.

SQL (RisingWave)

-- Step 1: the Kafka topic as a continuously read source.
-- Avro with Confluent Schema Registry; columns come from the schema.
CREATE SOURCE clicks
INCLUDE timestamp AS kafka_ts
WITH (
  connector = 'kafka',
  topic = 'web.clicks',
  properties.bootstrap.server = 'broker-1:9092,broker-2:9092',
  scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE AVRO (
  schema.registry = 'http://schema-registry:8081'
);

-- Reference data from Postgres via CDC, kept in sync automatically.
CREATE TABLE pages (
  page_id   BIGINT PRIMARY KEY,
  site      VARCHAR,
  category  VARCHAR
) WITH (
  connector = 'postgres-cdc',
  hostname = 'pg', port = '5432',
  username = 'rw', password = '...',
  database.name = 'cms', schema.name = 'public',
  table.name = 'pages'
);

Step 2: Transform with materialized views

Materialized views in RisingWave are maintained incrementally: each new Kafka message updates only the rows it affects. The rollup view reads from the clean view, so both stay consistent with each other. See how incremental materialized views work for the mechanics.

SQL (RisingWave)

-- Step 2: shape the data once, incrementally.
-- Dedup by event id, drop bots, enrich with page metadata.
CREATE MATERIALIZED VIEW clean_clicks AS
SELECT DISTINCT ON (c.event_id)
       c.event_id,
       c.user_id,
       c.page_id,
       p.site,
       p.category,
       c.event_ts
FROM clicks c
JOIN pages p ON c.page_id = p.page_id
WHERE c.user_agent NOT ILIKE '%bot%'
ORDER BY c.event_id, c.event_ts;

-- A second table from the same topic: one-minute rollups.
CREATE MATERIALIZED VIEW clicks_per_minute AS
SELECT window_start, site, category, COUNT(*) AS clicks
FROM TUMBLE(clean_clicks, event_ts, INTERVAL '1 minute')
GROUP BY window_start, site, category;

Step 3a: Sink into your existing catalog

Use this when Spark, Trino, Snowflake or Databricks already read from a Glue, REST, Hive or S3 Tables catalog and you want the new tables to appear next to the old ones. The first sink is append-only; the second is an upsert so that a late click revises its minute instead of adding a second row.

SQL (RisingWave)

-- Step 3a: sink into an existing Iceberg catalog (here: AWS Glue).
CREATE SINK clean_clicks_to_lake FROM clean_clicks
WITH (
  connector = 'iceberg',
  type = 'append-only',
  catalog.type = 'glue',
  catalog.name = 'lake',
  warehouse.path = 's3://my-lake/warehouse',
  database.name = 'web',
  table.name = 'clean_clicks',
  create_table_if_not_exists = 'true',
  partition_by = 'day(event_ts),bucket(16,user_id)',
  commit_checkpoint_interval = 60,
  auto.schema.change = 'true',
  s3.region = 'us-west-2',
  s3.access.key = '...',
  s3.secret.key = '...'
);

-- Upsert variant for the rollup: late events update the minute row.
CREATE SINK rollups_to_lake FROM clicks_per_minute
WITH (
  connector = 'iceberg',
  type = 'upsert',
  primary_key = 'window_start,site,category',
  write_mode = 'merge-on-read',
  catalog.type = 'glue',
  catalog.name = 'lake',
  warehouse.path = 's3://my-lake/warehouse',
  database.name = 'web',
  table.name = 'clicks_per_minute',
  create_table_if_not_exists = 'true',
  s3.region = 'us-west-2',
  s3.access.key = '...',
  s3.secret.key = '...'
);

Step 3b: Or let RisingWave manage the Iceberg table

With ENGINE = iceberg the table is a normal RisingWave table whose storage is Iceberg on your bucket, with RisingWave hosting the catalog. You can query it, index it, build views on it, and still read it from any other engine through the catalog. Compaction and snapshot expiration are switched on in the same statement.

SQL (RisingWave)

-- Step 3b (alternative): let RisingWave own the Iceberg table.
CREATE CONNECTION lake_conn WITH (
  type = 'iceberg',
  hosted_catalog = true,
  warehouse.path = 's3://my-lake/warehouse/',
  s3.region = 'us-west-2',
  s3.access.key = '...',
  s3.secret.key = '...'
);
SET iceberg_engine_connection = 'public.lake_conn';

CREATE TABLE clean_clicks_iceberg (
  event_id BIGINT PRIMARY KEY,
  user_id  BIGINT,
  page_id  BIGINT,
  site     VARCHAR,
  category VARCHAR,
  event_ts TIMESTAMPTZ
) WITH (
  commit_checkpoint_interval = 60,
  enable_compaction = true,
  enable_snapshot_expiration = true
) ENGINE = iceberg;

CREATE SINK fill_iceberg INTO clean_clicks_iceberg FROM clean_clicks;

-- Step 4: it is a normal Iceberg table. Query it here ...
SELECT site, COUNT(*) FROM clean_clicks_iceberg GROUP BY site;
-- ... or from Spark, Trino, DuckDB, Snowflake or Databricks
-- through the same catalog and warehouse path.

For a runnable local version of this pipeline with Kafka, Lakekeeper, MinIO and DuckDB in Docker Compose, see the Iceberg demo in awesome-stream-processing, and for AWS-native setups the 10-minute Iceberg quick start (built-in catalog or S3 Tables).

Decisions

Which settings decide whether the table stays fast?

Getting the first rows into Iceberg takes twenty minutes. Keeping the table fast after six months of continuous writes depends on six choices, and they apply whichever tool writes the table.

Append-only or upsert

Raw events are append-only. Anything with a primary key that can change (CDC tables, windowed aggregates that late data can revise) needs type = 'upsert' and a primary_key. Upsert writes Iceberg delete files, which readers must merge; that is the price of correctness for changing data.

Merge-on-read or copy-on-write

Merge-on-read (default) writes deletes as separate files and is cheapest for continuous ingestion. Copy-on-write rewrites data files on every update and is only available for upsert; choose it when the table is read far more than it is written and your readers are slow at applying delete files.

Commit interval

Each commit is a snapshot and at least one file per partition. Sixty seconds is a sensible default for analytics tables. Go lower only for tables where a human or an agent is waiting on the result, and pair it with compaction.

Partitioning

Use Iceberg hidden partitioning on the column readers filter by, usually a time column with day() or hour(), plus a bucket() on a high-cardinality key if queries are selective on it. Over-partitioning multiplies small files; a streaming writer should rarely have more than a few hundred live partitions.

Catalog

The catalog is what makes the table visible to other engines. Use Glue if your readers are on AWS, a REST catalog (Lakekeeper, Polaris) for a vendor-neutral setup, S3 Tables for an AWS-managed Iceberg service, or RisingWave's hosted catalog when RisingWave owns the table and others only read it.

Maintenance

Compaction merges small files; snapshot expiration deletes files no snapshot references. RisingWave can run both itself (enable_compaction, enable_snapshot_expiration, with a dedicated compactor node) or you can leave it to Spark or EMR. Without either, a streaming Iceberg table gets slower every day.

Reference: write modes, table maintenance, catalogs, Iceberg partition transforms.

Fit

When is RisingWave the wrong tool for Kafka to Iceberg?

→You need a byte-for-byte copy of a topic and nothing else. Kafka Connect, Tableflow or Iceberg Topics is less to run.
→Your freshness requirement is hours, not seconds or minutes. A scheduled Spark job is simpler and you probably already have the cluster.
→Your transformation needs custom operators that cannot be expressed in SQL or a UDF. Flink's DataStream API is built for that.

RisingWave earns its place when the pipeline has logic in it: deduplication, joins with CDC or lake tables, windowed aggregates, several output tables from one topic, or a requirement that the transformed result also be queryable in about a second before it reaches the lake.

Frequently Asked Questions

Kafka Connect Iceberg sink vs RisingWave: when is each the right choice?

If you only need a faithful copy of a topic in Iceberg and your schema is already how you want it, the Kafka Connect Iceberg sink (or a managed equivalent such as Confluent Tableflow or Redpanda Iceberg Topics) is the smallest moving part and you should use it. RisingWave is the right choice when the data has to change on the way: deduplication, joining with reference tables or CDC streams, aggregating into windows, or producing several differently shaped tables from one topic. Those steps otherwise require a separate Flink or Spark job.

How does RisingWave guarantee exactly-once delivery to Iceberg?

The Iceberg sink commits a new table snapshot at RisingWave checkpoints. Each commit is tied to a checkpoint epoch and recorded so that, after a failure, RisingWave can tell which checkpoints have already been committed and will not commit them twice. Exactly-once is on by default (is_exactly_once = true) and requires sink decoupling, which is also the default. Turning it off trades duplicates on recovery for slightly lower overhead.

What end-to-end latency should I expect from Kafka to a queryable Iceberg row?

The dominant term is the commit interval. RisingWave commits to Iceberg every commit_checkpoint_interval checkpoints; with the one-second default checkpoint and the default value of 60, a row is visible to Iceberg readers about a minute after it was read from Kafka. You can lower the interval to a few seconds for demos or latency-sensitive tables, at the cost of more and smaller files and more snapshots. Inside RisingWave the transformed result is queryable in about a second regardless of the Iceberg commit interval.

How do I avoid the small files problem when streaming into Iceberg?

Every commit writes at least one data file per partition and per writer, so a streaming writer produces small files by design. Two controls matter. First, do not commit more often than your readers need (sixty seconds is a reasonable default). Second, enable compaction so files are merged in the background: RisingWave's own compactor (enable_compaction = true on the sink or table, with a dedicated compactor node) or an external compactor such as Spark or Amazon EMR. Snapshot expiration should run alongside compaction so old files are actually deleted.

Can I write updates and deletes, for example from a Debezium or compacted topic?

Yes. Create the sink or table with type = 'upsert' and a primary_key. RisingWave then writes Iceberg equality deletes in merge-on-read mode (the default), or rewrites data files in copy-on-write mode if you prefer faster reads over cheaper writes. Append-only sinks must use merge-on-read. If your Kafka messages are Debezium envelopes, use FORMAT DEBEZIUM on the source so RisingWave interprets inserts, updates and deletes correctly.

Does RisingWave work with Confluent Schema Registry and Avro?

Which Iceberg catalogs and object stores are supported?

Catalogs: AWS Glue, REST (including Lakekeeper, Polaris and other REST implementations), Hive Metastore, JDBC, Amazon S3 Tables, and RisingWave's own hosted catalog for RisingWave-managed tables. Object storage: Amazon S3, Google Cloud Storage, Azure Blob and S3-compatible stores such as MinIO. The catalog decides which external engines can see the table, so pick the one your Spark, Trino, Snowflake or Databricks users already use.

Can RisingWave also read Iceberg tables back?

Stream your first topic into Iceberg

One source, one view, one sink. Query the table from RisingWave and from your lake engines.

Start Free

Iceberg Streaming →Materialized Views on Iceberg →RisingWave and Kafka →