Real-Time Lakehouse with RisingWave and Apache Iceberg: Ingestion, Querying, and Architecture

A real-time lakehouse with RisingWave and Apache Iceberg replaces the Debezium, Kafka, Flink, and Airflow chain with one PostgreSQL-compatible streaming database that ingests CDC and Kafka events, transforms them in SQL materialized views, commits results to Iceberg tables on S3 every few seconds with built-in compaction, and reads Iceberg tables back as sources to join history with live streams. This guide covers the architectural choice between a multi-component pipeline and a single system, then builds the full pipeline: Kafka and PostgreSQL CDC ingestion into Iceberg, catalog and write-mode options, S3 layout and cost, querying Iceberg from Trino, Athena, and RisingWave itself, and the lakehouse join patterns that batch engines cannot express.

What a Real-Time Lakehouse Is

A data lakehouse merges the warehouse's strong consistency, structured schemas, and query performance with the data lake's open formats and object storage economics. Amazon S3 stores data at roughly $0.023 per GB per month, 10 to 100 times cheaper than the $5 to $50 per TB per month typical of managed warehouses, and Apache Iceberg adds ACID transactions, snapshot isolation, partition pruning, schema evolution, and time travel on top of it, so any engine, Trino, Athena, Spark, DuckDB, or Snowflake, reads the same tables.

The real-time qualifier adds a freshness requirement. A traditional lakehouse pipeline runs hourly Spark jobs from Kafka to Iceberg and analysts query data that is one to two hours stale. For dashboards tracking sales, fraud, engagement, or operations, that is the gap. Freshness measured in seconds is what forces the architectural decision, because batch ETL cannot meet it.

Architecture 1: The Multi-Component Pipeline

The dominant production pattern assembles best-of-breed components: Debezium or a managed CDC service reads the database write-ahead log and emits row-level changes; Kafka buffers them and provides replay; Flink or Spark Structured Streaming consumes topics, transforms, and writes downstream; Iceberg receives committed data and serves readers; Trino, Athena, or Spark SQL runs ad hoc queries; Airflow schedules compaction, snapshot expiration, and orphan file cleanup; and Prometheus and Grafana monitor each component separately. That is five to seven components, each independently scalable and deployable, with mature tooling, large communities, and documented playbooks.

[Source DB] --WAL/CDC--> [Debezium] --> [Kafka] --> [Flink] --> [Iceberg on S3] --> [Trino / Athena]
                                                                       ^
                                                              [Airflow compaction]

Architecture 2: The Single-System Pipeline

RisingWave collapses the ingestion and transformation layers. It connects directly to MySQL, PostgreSQL, MongoDB, and other sources through built-in CDC connectors, or consumes Kafka topics when an event bus already exists upstream, defines transformations as SQL materialized views, and writes results to Iceberg through a native sink that handles compaction. The query engine still reads Iceberg. The component count drops from seven to four, and the ingestion path itself is three: source, RisingWave, Iceberg. RisingWave is open source under Apache 2.0, written in Rust, and separates compute from state storage on S3 so each scales independently.

[Source DB] --CDC or Kafka--> [RisingWave: SQL materialized views] --> [Iceberg on S3] --> [Trino / Athena / RisingWave]

Consistency Is the Underestimated Problem

In the multi-component design, data exists in three states at once: raw events in Kafka, intermediate state in Flink checkpoints, and committed rows in Iceberg, maintained by three systems with three failure and recovery models. If a Flink job crashes between a checkpoint and an Iceberg commit, recovery replays Kafka from the last checkpoint offset and re-attempts the write; if Kafka retention expired or the checkpoint is corrupt, data is lost. In the single-system design, RisingWave owns the path from ingestion to Iceberg commit, and its internal checkpoint and the Iceberg commit are coordinated by the same engine. Failure modes still exist, but the failure domain is one system rather than three.

Schema Evolution

Adding a column at the source in the multi-component design propagates through the CDC connector configuration, the Schema Registry subject, the Flink job's type definitions, and the Iceberg table schema, each with its own tooling and failure modes. RisingWave detects the change on the CDC stream and uses Iceberg's native schema evolution to propagate it.

When Multi-Component Still Makes Sense

Petabyte-scale throughput above hundreds of terabytes a day, where Flink's fine-grained resource partitioning is hard to match in one system. Mixed team ownership, where the Kafka team, the transformation team, and the lakehouse team are different organizations and component boundaries map to org boundaries. Years of existing Flink expertise, custom connectors, and tooling whose switching cost exceeds the simplification. Compliance frameworks that require physical boundaries between processing stages. For a new project at moderate scale, up to tens of terabytes a day, the single system is the faster path to production with a smaller operational burden.

Ingestion Path 1: Kafka Events to Iceberg

Three statements: a source, an optional materialized view, a sink. No JVM, no cluster, no scheduled compaction job. The event stream is append-only, so the sink is append-only.

CREATE SOURCE app_events (
    event_id     VARCHAR,
    user_id      BIGINT,
    session_id   VARCHAR,
    event_name   VARCHAR,
    page_path    VARCHAR,
    revenue      NUMERIC(10,2),
    device_type  VARCHAR,
    geo_country  VARCHAR,
    event_time   TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'app-events', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'earliest')
FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW app_events_enriched AS
SELECT event_id, user_id, session_id, event_name, page_path, revenue, device_type, geo_country, event_time,
       event_time::DATE AS event_date, EXTRACT(HOUR FROM event_time) AS event_hour
FROM app_events;

CREATE SINK app_events_to_iceberg AS SELECT * FROM app_events_enriched
WITH (
    connector      = 'iceberg',
    type           = 'append-only',
    catalog.type   = 'rest',
    catalog.uri    = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://analytics-lake/warehouse',
    s3.region      = 'us-east-1',
    database.name  = 'analytics',
    table.name     = 'app_events'
);

Aggregations go the same way, in upsert mode because a window's row changes as events arrive. Keep metrics at several granularities: five-minute tumbling windows for the live dashboard, hourly for trends, and a one-hour HOP sliding every five minutes for "last hour" tiles.

CREATE MATERIALIZED VIEW realtime_5min_metrics AS
SELECT geo_country, device_type, window_start, window_end,
       COUNT(DISTINCT user_id) AS active_users, COUNT(DISTINCT session_id) AS active_sessions,
       COUNT(*) FILTER (WHERE event_name = 'purchase') AS purchase_events,
       SUM(revenue) FILTER (WHERE event_name = 'purchase') AS revenue_5min, COUNT(*) AS total_events
FROM TUMBLE(app_events, event_time, INTERVAL '5 MINUTES')
GROUP BY geo_country, device_type, window_start, window_end;

CREATE MATERIALIZED VIEW hourly_analytics AS
SELECT geo_country, device_type, window_start AS hour_bucket,
       COUNT(DISTINCT user_id) AS unique_users, COUNT(DISTINCT session_id) AS unique_sessions, SUM(revenue) AS total_revenue,
       COUNT(*) FILTER (WHERE event_name = 'page_view') AS page_views, COUNT(*) FILTER (WHERE event_name = 'add_to_cart') AS add_to_cart_events,
       COUNT(*) FILTER (WHERE event_name = 'purchase') AS purchases, AVG(revenue) FILTER (WHERE event_name = 'purchase') AS avg_order_value
FROM TUMBLE(app_events, event_time, INTERVAL '1 HOUR')
GROUP BY geo_country, device_type, window_start;

CREATE MATERIALIZED VIEW rolling_1hr_active_users AS
SELECT geo_country, window_start, window_end, COUNT(DISTINCT user_id) AS active_users_1hr, SUM(revenue) AS rolling_revenue_1hr
FROM HOP(app_events, event_time, INTERVAL '5 MINUTES', INTERVAL '1 HOUR')
GROUP BY geo_country, window_start, window_end;

CREATE SINK hourly_analytics_to_iceberg AS SELECT * FROM hourly_analytics
WITH (
    connector      = 'iceberg',
    type           = 'upsert',
    primary_key    = 'geo_country,device_type,hour_bucket',
    catalog.type   = 'rest',
    catalog.uri    = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://analytics-lake/warehouse',
    s3.region      = 'us-east-1',
    database.name  = 'analytics',
    table.name     = 'hourly_analytics'
);

Athena and Trino see data at most one checkpoint interval old, typically 10 to 60 seconds. For sub-second freshness, BI tools connect to RisingWave directly over the PostgreSQL protocol and read the materialized views before the Iceberg commit. The dual-serving pattern, views for live queries and Iceberg for history, is the common end state.

Ingestion Path 2: PostgreSQL CDC to Iceberg

RisingWave's postgres-cdc connector reads the logical replication slot directly, so the CDC path needs no Kafka and no Debezium deployment, with sub-second replication lag and exactly-once delivery backed by Iceberg's atomic commits. PostgreSQL needs wal_level = logical, enough replication slots and WAL senders, and a replication user; pgoutput ships with PostgreSQL 10 and later.

-- On PostgreSQL
CREATE USER cdc_user REPLICATION LOGIN PASSWORD 'secure_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;

-- In RisingWave: one shared CDC source, one table per upstream table
CREATE SOURCE pg_ecommerce WITH (
    connector        = 'postgres-cdc',
    hostname         = 'postgres.prod.internal',
    port             = '5432',
    username         = 'cdc_user',
    password         = 'secure_password',
    database.name    = 'ecommerce',
    schema.name      = 'public',
    slot.name        = 'risingwave_slot',
    publication.name = 'risingwave_publication'
);

CREATE TABLE products (
    product_id      BIGINT PRIMARY KEY,
    sku             VARCHAR,
    name            VARCHAR,
    category        VARCHAR,
    price           NUMERIC(12,2),
    stock_quantity  INT,
    is_active       BOOLEAN,
    created_at      TIMESTAMPTZ,
    updated_at      TIMESTAMPTZ
) FROM pg_ecommerce TABLE 'public.products';

CREATE TABLE orders (
    order_id     BIGINT PRIMARY KEY,
    customer_id  BIGINT,
    product_id   BIGINT,
    quantity     INT,
    unit_price   NUMERIC(12,2),
    status       VARCHAR,
    ordered_at   TIMESTAMPTZ,
    shipped_at   TIMESTAMPTZ
) FROM pg_ecommerce TABLE 'public.orders';

RisingWave creates the slot and the publication and takes an initial snapshot before switching to streaming. Raw CDC rows usually need enrichment before landing in the lakehouse, and the enrichment is a join that RisingWave maintains as either side changes.

CREATE TABLE category_metadata (
    category     VARCHAR PRIMARY KEY,
    department   VARCHAR,
    margin_pct   NUMERIC(5,2),
    is_seasonal  BOOLEAN
);

CREATE MATERIALIZED VIEW products_enriched AS
SELECT p.product_id, p.sku, p.name, p.category, c.department, c.margin_pct, p.price,
       p.price * (1 - c.margin_pct / 100) AS cost_basis, p.stock_quantity, p.is_active, p.created_at, p.updated_at
FROM products p
LEFT JOIN category_metadata c ON p.category = c.category;

CREATE MATERIALIZED VIEW orders_with_products AS
SELECT o.order_id, o.customer_id, o.quantity, o.unit_price, o.status, o.ordered_at, o.shipped_at, p.sku, p.category, p.department
FROM orders o
JOIN products_enriched p ON o.product_id = p.product_id;

CREATE SINK products_iceberg_sink AS SELECT * FROM products_enriched
WITH (connector = 'iceberg', type = 'upsert', primary_key = 'product_id', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
      warehouse.path = 's3://ecommerce-lakehouse/warehouse', s3.region = 'us-east-1', database.name = 'ecommerce', table.name = 'products');

CREATE SINK orders_iceberg_sink AS SELECT * FROM orders_with_products
WITH (connector = 'iceberg', type = 'upsert', primary_key = 'order_id', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
      warehouse.path = 's3://ecommerce-lakehouse/warehouse', s3.region = 'us-east-1', database.name = 'ecommerce', table.name = 'orders');

When a product's price changes in PostgreSQL, the view updates immediately and the sink writes an equality delete for the old row and a new data record in the same Iceberg snapshot. Deletes in the source propagate as deletes. Replication health is visible on both sides:

-- RisingWave
SELECT name, connector, status FROM rw_sources WHERE connector = 'postgres-cdc';

-- PostgreSQL
SELECT slot_name, confirmed_flush_lsn, pg_current_wal_lsn(),
       pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes
FROM pg_replication_slots
WHERE slot_name LIKE 'risingwave_%';

Steady-state lag is under a second for moderate workloads. Watch lag_bytes on the PostgreSQL side: a slot whose consumer has stopped retains WAL indefinitely and eventually fills the disk.

Write Modes, Catalogs, and Compaction

Append-only sinks fit event streams, logs, clickstream, and IoT, where rows are never updated. Upsert sinks with a primary_key fit CDC and aggregations; they default to merge-on-read for write throughput and can be configured copy-on-write for read performance. The REST catalog is the recommended choice and works with Polaris, Lakekeeper, Tabular, Nessie, and Unity Catalog's Iceberg REST endpoint; Hive Metastore, AWS Glue, AWS S3 Tables, and JDBC catalogs are also supported.

-- REST (recommended)
WITH (connector = 'iceberg', catalog.type = 'rest', catalog.uri = 'http://polaris:8181', ...)
-- Hive Metastore
WITH (connector = 'iceberg', catalog.type = 'hive', catalog.uri = 'thrift://hive-metastore:9083', ...)
-- AWS S3 Tables
WITH (connector = 'iceberg', catalog.type = 's3_tables', catalog.name = 'my-s3-table-bucket', s3.region = 'us-east-1', ...)
-- JDBC
WITH (connector = 'iceberg', catalog.type = 'jdbc', catalog.uri = 'jdbc:postgresql://catalog-db:5432/iceberg', catalog.jdbc.user = 'iceberg', catalog.jdbc.password = '...', ...)

Streaming writes produce many small Parquet files, typically 1 to 10 MB each, and Iceberg queries scan every file matching the partition filter, so the small-files problem degrades reads unless something rewrites them. In the multi-component design that is an Airflow-scheduled Spark RewriteDataFiles or PyIceberg rewrite_data_files job with its own monitoring. RisingWave's Iceberg sink compacts automatically, so the compaction job, its schedule, and its alerting disappear from the stack. Partition the Iceberg tables by the dimensions queries filter on, date and country for the analytics tables above, so a query for one country on one day scans one partition.

S3 Layout, Lifecycle, and IAM

Organize the warehouse so lifecycle policies apply cleanly, one prefix per database and table with Iceberg's own metadata and data directories under each. Streaming data is read heavily for about 30 days and rarely after, so tier it:

Prefer IAM roles over access keys: on EC2 or EKS, attach a role and omit s3.access.key and s3.secret.key from the sink. The role needs s3:GetObject, s3:PutObject, s3:DeleteObject, s3:ListBucket, and s3:GetBucketLocation on the bucket and its objects, plus glue:GetDatabase, glue:GetTable, glue:CreateTable, and glue:UpdateTable when Glue is the catalog.

Querying the Lakehouse: Batch Engines and RisingWave as a Reader

Athena and Trino query the tables as soon as a snapshot commits:

-- Athena, Iceberg table registered in Glue
SELECT account_id, SUM(gross_volume) AS total_volume, SUM(failed_count) AS total_failures
FROM finance.account_daily_summary
WHERE txn_date >= DATE '2026-01-01'
GROUP BY account_id
ORDER BY total_volume DESC
LIMIT 100;

Batch engines cannot run a continuous query, join a live stream against a table, or compute incrementally; every query rescans. RisingWave registers Iceberg tables as sources and builds materialized views on them, reading only the new files when a new snapshot commits, whether RisingWave's own sink or a nightly Spark job wrote it.

CREATE SOURCE orders_iceberg WITH (
    connector = 'iceberg', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://ecommerce-lakehouse/warehouse', s3.region = 'us-east-1', database.name = 'ecommerce', table.name = 'orders'
);

SELECT department, COUNT(*) AS order_count, SUM(quantity * unit_price) AS revenue
FROM orders_iceberg
WHERE ordered_at >= '2026-01-01'
GROUP BY department
ORDER BY revenue DESC
LIMIT 20;

CREATE MATERIALIZED VIEW iceberg_daily_revenue AS
SELECT department, window_start::DATE AS day, COUNT(*) AS orders, SUM(quantity * unit_price) AS revenue
FROM TUMBLE(orders_iceberg, ordered_at, INTERVAL '1 DAY')
GROUP BY department, window_start;

Joining Live Streams with Lakehouse History

The pattern no batch engine offers: enrich live events with historical context from Iceberg, or compare the live hour to the same hour last week, in one view with no data movement.

CREATE SOURCE user_profiles_iceberg WITH (
    connector = 'iceberg', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://analytics-lake/warehouse', s3.region = 'us-east-1', database.name = 'users', table.name = 'user_profiles'
);

CREATE MATERIALIZED VIEW enriched_live_events AS
SELECT e.user_id, e.event_name, e.page_path, e.event_time, p.user_segment, p.account_age_days, p.lifetime_value, p.preferred_language
FROM app_events e
LEFT JOIN user_profiles_iceberg p ON e.user_id = p.user_id;

CREATE SOURCE iceberg_hourly_historical WITH (
    connector = 'iceberg', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://analytics-lake/warehouse', s3.region = 'us-east-1', database.name = 'analytics', table.name = 'hourly_analytics'
);

CREATE MATERIALIZED VIEW revenue_week_over_week AS
SELECT live.geo_country, live.hour_bucket, live.total_revenue AS current_revenue, hist.total_revenue AS last_week_revenue,
       ROUND(100.0 * (live.total_revenue - hist.total_revenue) / NULLIF(hist.total_revenue, 0), 1) AS revenue_growth_pct
FROM hourly_analytics live
JOIN iceberg_hourly_historical hist ON live.geo_country = hist.geo_country AND hist.hour_bucket = live.hour_bucket - INTERVAL '7 DAYS'
WHERE live.total_revenue > 0;

CREATE SOURCE batch_user_features WITH (
    connector = 'iceberg', catalog.type = 'rest', catalog.uri = 'http://iceberg-catalog:8181',
    warehouse.path = 's3://ml-lake/warehouse', s3.region = 'us-east-1', database.name = 'ml_features', table.name = 'user_features'
);

CREATE MATERIALIZED VIEW scoring_input AS
SELECT e.user_id, e.session_id, e.event_name, f.feature_vector_json, f.risk_score, f.last_model_update
FROM app_events e
JOIN batch_user_features f ON e.user_id = f.user_id;

Feature enrichment for online scoring, anomaly detection against historical baselines, trend comparison, backfill from raw history, and cross-table reporting all follow one of these shapes. Four habits keep Iceberg reads fast: include partition key filters so predicates push down to the manifest scan, select only the columns needed so projection happens at the Parquet level, add a date filter when building a view over a large table so startup does not scan years, and set a short snapshot polling interval when freshness matters. Time travel to a specific snapshot or timestamp remains a Trino or Spark FOR SYSTEM_VERSION AS OF query; RisingWave reads the current snapshot.

Multi-Component vs Single System

Frequently Asked Questions

Do I still need Kafka? Not for the CDC path; postgres-cdc and mysql-cdc read the log directly. Keep Kafka where it already serves as the event bus for other consumers, and let RisingWave consume it without a Flink layer in between.

How fresh is data in Iceberg? One checkpoint interval, typically 10 to 60 seconds, configurable per sink. Queries that need sub-second freshness read RisingWave's materialized views directly over PostgreSQL and the same data lands in Iceberg moments later.

How are updates and deletes represented in Iceberg? Upsert sinks write equality delete files for the old row and a data file for the new one in the same snapshot, so readers see a consistent table. Merge-on-read is the default; copy-on-write is available when read latency matters more than write throughput.

Which catalogs and engines work? REST catalogs (Polaris, Lakekeeper, Tabular, Nessie, Unity), Hive Metastore, AWS Glue, S3 Tables, and JDBC. Any engine that reads Iceberg, Trino, Athena, Spark, DuckDB, Snowflake, BigQuery, reads the tables RisingWave writes.

Can RisingWave read Iceberg tables written by other engines? Yes. Register the table as a source and RisingWave reads new snapshots incrementally, whoever committed them. That makes nightly Spark output, such as ML feature tables, joinable with live Kafka events in a single materialized view.