Streaming Into the Lakehouse: Kafka, Flink and Apache Iceberg in Real Time

The two-copies problem

In most estates we assess, the same order, claim or payment event exists twice. The streaming copy lives in Kafka and drives applications, fraud checks and notifications within seconds. The analytics copy lives in the warehouse or lakehouse and is built by a nightly or hourly batch job that reads from the source database, from Kafka, or from files someone dropped in a bucket.

The two copies drift. They have different schemas, different definitions of “late”, different handling of deletes and different owners. When the fraud dashboard and the finance report disagree, two teams spend a week proving whose pipeline is right. And every new AI use case — a feature store, a retrieval index, an agent that needs current context — asks for a third copy.

Apache Iceberg changes the economics. It is an open table format: data files in Parquet (or ORC/Avro) on object storage, plus metadata that tracks which files make up each snapshot of the table. Any engine that speaks Iceberg can read the same table with the same schema and the same transactional guarantees. If the stream writes that table continuously, the analytics copy stops being a separate system and becomes a view of the stream with a longer memory.

The reference architecture

The pattern has five layers. Producers and CDC connectors write to Kafka with schemas enforced by a schema registry. An ingestion layer turns topics into Iceberg commits. A catalog records which metadata file is current for each table. Maintenance jobs keep the table healthy. Query engines and downstream consumers read it.

Two properties make this work. First, an Iceberg commit is an atomic swap of one metadata pointer in the catalog: readers see the whole commit or none of it, never a half-written batch. Second, the table carries its own schema and partition spec, so every engine reads it the same way. Those are the two things the batch copy never gave you.

Three ways to land a topic as a table

There are three common ingestion paths, and most estates end up using more than one.

Kafka Connect Iceberg sink

The Apache Iceberg project ships a Kafka Connect sink connector. Workers read topic partitions, write Parquet data files, and a coordinator commits the files from all workers to the table at a configured interval. It is the shortest path for “land this topic as a table”: configuration rather than code, fan-out to many tables from one connector, and schema evolution driven by the record schema. Its limits show up when you need joins, enrichment, deduplication across keys or complex upsert logic — that is processing, and Connect is a movement tool.

Flink with the Iceberg connector

A Flink job reads from Kafka, transforms, and writes through the Iceberg sink. It is the right tool when the table is not a raw copy of the topic: when you need to clean, enrich, join, aggregate, deduplicate or apply CDC upserts by primary key. Flink ties Iceberg commits to its own checkpoints, which is what gives you exactly-once (more on that below). The cost is that you now operate a stateful stream-processing job, with checkpoint storage, state sizing and upgrade procedures.

Managed topic-to-table features

Several streaming platforms and lakehouse vendors now materialize topics as Iceberg tables directly, without a connector or job for you to run. They are attractive for raw and bronze layers: no infrastructure, consistent naming, schemas taken from the registry. Check three things before you rely on one: which catalog it commits to and whether your other engines can use that catalog, how it handles deletes and upserts, and who owns compaction and snapshot expiry on the resulting tables.

Exactly-once into object storage

Object storage has no transactions, and Kafka offsets live somewhere else. “Exactly-once” into a lake therefore depends entirely on one question: do the consumed offsets and the data become durable in the same atomic step?

With Flink, the answer can be yes. The Iceberg sink writes data files continuously, but it only commits them when a checkpoint completes. The checkpoint records the Kafka offsets and the list of files written since the last checkpoint. If the job fails, it restarts from the last completed checkpoint, re-reads from those offsets, and files from the failed interval are never committed. Iceberg also records the checkpoint ID in the snapshot summary, so a commit that already succeeded is not applied twice on recovery.

The Iceberg Connect sink uses a similar idea: a coordinator collects the files written by all workers and commits them together, and the committed offsets are tracked alongside the table commit so a restart does not double-write.

The guarantee gap with plain file sinks

Generic “topic to S3” file sinks are a different story. They write Parquet or JSON objects to a path, and readers discover data by listing files. There is no table commit, so the guarantees depend on details: whether file names are deterministic per offset range so a retry overwrites instead of duplicating, whether a reader can see a file that a crashed task half-finished, and whether any downstream job dedups. Many teams believe they have exactly-once because the connector documentation says so for one configuration, and lose it when they change partitioning, rotation or the writer. If the destination is an analytics table, land it as a table with commits, not as a folder of files.

The commit interval: freshness versus everything else

Every commit creates at least one new data file per writer and partition, a manifest, a manifest list and a metadata file, and makes one call to the catalog. Commit every few seconds and the table is fresh but full of tiny files and very long snapshot history. Commit every hour and the files are well sized but the table is an hourly batch again.

The right answer is per table, agreed with the people who read it. A useful question to ask them: “What decision gets worse if this table is five minutes old instead of thirty seconds?” If there is no answer, five minutes is fine. If there is, the decision probably belongs on the stream, not on the table.

CDC upserts: deletes, merges and the v3 changes

Append-only events are easy: every record becomes a row. Change data capture is harder, because each record is an insert, update or delete of a row identified by a primary key, and the table must reflect the latest state.

Iceberg format version 2 introduced row-level deletes with two kinds of delete files:

• Equality deletes say “delete every row where id = 42”. They are cheap to write from a stream, because the writer does not need to know where the old row lives. They are expensive to read, because every reader must apply them against data files until they are compacted away. This is what streaming upsert writers such as the Flink sink typically produce.
• Position deletes say “delete row 1,207 of file X”. They are cheap to read but require the writer to locate the row first, which batch engines such as Spark do during a MERGE.

Format version 3 adds deletion vectors: compact bitmaps, stored in Puffin files, that mark deleted row positions for a data file, with at most one deletion vector per data file. They replace position delete files for v3 tables and make merge-on-read considerably cheaper for readers. Puffin is also where Iceberg keeps auxiliary statistics such as column sketches that engines can use for planning. Engine support for v3 is still rolling out, so check every engine that reads the table before upgrading its format version — a writer on v3 and a reader that only understands v2 is a broken table for that reader.

The table write mode decides who pays:

Two practical rules for CDC tables. First, key the Kafka topic by the table’s primary key, so all changes for a row arrive in order on one partition; otherwise an older update can overwrite a newer one. Second, keep an append-only change log table next to the current-state table when audit or replay matters. The current-state table answers “what is true now”; the change log answers “what happened and when”, and it is much easier to rebuild the first from the second than the other way round.

Table maintenance is part of the pipeline

The most common reason streaming Iceberg projects disappoint is not ingestion. It is that nobody owns the table after the data lands. A streaming table needs four jobs, run on a schedule and monitored like any other production pipeline:

Many catalogs and platforms now offer managed maintenance for some or all of these. Use it when it fits, but confirm which tables it covers, how it is configured, and whether it is aware of your streaming writer.

Partitioning, schemas and catalogs

Partition for how the table is read

Iceberg’s hidden partitioning lets you partition by a transform of a column — days(event_ts), bucket(16, customer_id) — without readers needing to know the scheme or filter on a derived column. Partition by event time for event tables, and keep partitions coarse enough that each one receives well-sized files between compactions. Over-partitioning a streaming table, for example by hour and by a high-cardinality key, multiplies the small-file problem by the number of partitions. Because partition specs can evolve, you can start coarse and refine later without rewriting history.

Keep the table schema and the topic schema in step

The schema registry is the contract for the topic; the Iceberg schema is the contract for the table. Align their evolution rules. Additive changes — a new optional field — should flow through automatically. Breaking changes — a type change, a removed required field, a renamed key — should be blocked at the registry by compatibility settings, not discovered when the sink fails at 2 a.m. Iceberg tracks columns by ID, so renames and reorders are safe in the table; make sure your sink maps fields by name or ID consistently.

Choose the catalog for the engines you have

The catalog decides who can commit and which engines can read. The Iceberg REST catalog specification has become the common interface, and most options now speak it: Apache Polaris, Databricks Unity Catalog, AWS Glue, Project Nessie and others, plus vendor-managed catalogs. Choose based on where access control is enforced, which engines need to write versus only read, multi-region needs, and whether the catalog vends short-lived storage credentials so engines never hold broad bucket keys. On Databricks estates we often use Unity Catalog as the governance plane, as in our Databricks POD reference designs; elsewhere a REST catalog shared by Spark, Trino and Flink is the usual pattern.

Lineage from topic to dashboard

When the stream and the table are one pipeline, lineage should be one graph too. OpenLineage gives Flink, Spark, Airflow and other tools a shared event model for “this job read these datasets and wrote these”. Emitting it from the ingestion layer, the maintenance jobs and the downstream transformations lets you trace a dashboard figure back to the topic and the source table — and know which tables are affected when a producer changes a schema.

A minimal Flink example

The shape of a Flink SQL pipeline from a Kafka topic to an Iceberg table, using a REST catalog. Connection details and options vary by version and platform.

-- 1. Register the Iceberg catalog (REST)
CREATE CATALOG lake WITH (
  'type' = 'iceberg',
  'catalog-type' = 'rest',
  'uri' = 'https://catalog.example.internal/api/catalog',
  'warehouse' = 'analytics'
);

-- 2. Read the Kafka topic with its registry schema
CREATE TABLE orders_src (
  order_id STRING, customer_id STRING, status STRING,
  amount DECIMAL(12,2), event_ts TIMESTAMP_LTZ(3),
  WATERMARK FOR event_ts AS event_ts - INTERVAL '30' SECOND
) WITH (
  'connector' = 'kafka', 'topic' = 'orders.v1',
  'properties.bootstrap.servers' = 'broker:9092',
  'properties.group.id' = 'orders-to-iceberg',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'avro-confluent',
  'avro-confluent.url' = 'https://registry.example.internal'
);

-- 3. Upsert target, partitioned by day, keyed by order_id
CREATE TABLE lake.sales.orders (
  order_id STRING, customer_id STRING, status STRING,
  amount DECIMAL(12,2), event_ts TIMESTAMP_LTZ(3), event_date DATE,
  PRIMARY KEY (order_id, event_date) NOT ENFORCED
) PARTITIONED BY (event_date)
  WITH ('format-version' = '2', 'write.upsert.enabled' = 'true');

-- 4. Commits happen on each checkpoint (e.g. every 2 minutes)
SET 'execution.checkpointing.interval' = '2 min';
INSERT INTO lake.sales.orders
SELECT order_id, customer_id, status, amount, event_ts, CAST(event_ts AS DATE)
FROM orders_src;

Note the partition column in the primary key: an upsert table needs the partition source inside the identifier so an update finds its row in the right partition. In practice we also add a dedup step for at-least-once producers, and a separate append-only change log table.

One table, many engines

The payoff is that the same table serves several consumers without copies. Spark and Databricks run heavy transformations and ML feature pipelines. Trino serves interactive SQL. Snowflake can query Iceberg tables in place through its catalog integrations. Flink itself can read the table incrementally for downstream streaming jobs, or in batch mode to backfill. Each engine has its own level of support for row-level deletes and v3 features, so we keep a short compatibility matrix per table: which engines read it, which version and features they need, and which one is allowed to write.

The same table also becomes the source for AI work: feature computation, embedding refresh for retrieval, and point-in-time training sets via snapshot time travel. That is where the two-copies problem was costing the most, because every AI team was building its own extract.

The failure modes we plan for

Readiness checklist

The bottom line

Kafka and the lakehouse no longer need to be two worlds with a nightly bridge between them. With Iceberg as the table format, the stream can write the analytics table directly, every engine can read it, and the two-copies problem goes away.

The work is in the decisions around the write: the right ingestion path per table, a commit interval chosen for real readers, CDC that respects key ordering and delete semantics, exactly-once mapped hop by hop, and table maintenance that ships with the pipeline. Get those right and the lakehouse becomes the long memory of the stream — governed, queryable from anywhere, and ready for the AI work that has been waiting for one trustworthy copy of the data.