CDC & lake sinks — Debezium changelog to Iceberg, commit on checkpoint, schema evolution

6 h
requires FLK-04
requires FLK-06

From your world. loading your lakehouse: a CDC feed from an operational database landing as an upserted Iceberg table your dbt models read

← All exercises

Apply a Debezium changelog into an upserted Iceberg table

Build the SQL for a CDC → Iceberg pipeline. The test sets up the infrastructure. It creates an Iceberg hadoop catalog on a local file:// warehouse. It also creates a debezium-json source table over a fixed file of change events. You write the pieces that make it an upsert and that evolve the schema. Then the test reconciles the Iceberg table against an independent DuckDB fold of the same change events (G6). It never compares against your SQL text.

The change events (phase 1)

c  id=1  name=alice  amt=100
c  id=2  name=bob    amt=50
u  id=1  name=alice  amt=175
d  id=2

Final current state after phase 1: only id=1 with amt=175.

The change events (phase 2, after schema evolution)

A new column region appears in the source:

c  id=3  name=carol  amt=70   region=EU
u  id=1  name=alice  amt=120  region=US

Final current state after phase 2: id=1 (amt=120, region=US) and id=3 (carol, 70, EU). Rows that existed before the column was added read NULL for region.

What to implement in `cdc_to_iceberg.py`

target_ddl(table: str) -> str — Return the CREATE TABLE <table> (…) for the Iceberg sink. It must be upsert-capable. It has the columns id INT, name STRING, amt INT, a PRIMARY KEY (id) NOT ENFORCED, and WITH ('format-version'='2'). The test does not add IF NOT EXISTS for you. Return a plain CREATE TABLE.
upsert_sql(src: str, tgt: str) -> str — Return INSERT INTO <tgt> SELECT id, name, amt FROM <src>. It applies the changelog (the src table already emits +I/-U/+U/-D).
evolve_ddl(table: str) -> str — Return the ALTER TABLE <table> ADD (region STRING) that evolves the schema.
upsert_sql_v2(src: str, tgt: str) -> str — Return the phase-2 upsert that also carries the new column: INSERT INTO <tgt> SELECT id, name, amt, region FROM <src>.

Constraints

Return SQL strings. The test runs them in the right order on the embedded Flink flink_sql runner.
Use the parameters (table, src, tgt) that you get for the identifiers.
The test asserts three things. (a) The table reconciles with the DuckDB fold after each phase. (b) At least one Iceberg snapshot was committed (commit-on-checkpoint). (c) After evolution, old rows read NULL for region. The test grades nothing on SQL text or timing.

Starter code

cdc_to_iceberg.py

"""Starter — make this an UPSERT pipeline. The stubs below create an append-only
table and never carry the new column, so the table does not reconcile with the
source (G1 fails) until you add the primary key + format-version 2 and the
evolution. See README.
"""


def target_ddl(table: str) -> str:
    # TODO: make this upsert-capable (PRIMARY KEY + format-version 2).
    return f"CREATE TABLE {table} (id INT, name STRING, amt INT)"


def upsert_sql(src: str, tgt: str) -> str:
    return f"INSERT INTO {tgt} SELECT id, name, amt FROM {src}"


def evolve_ddl(table: str) -> str:
    # TODO: ALTER TABLE ... ADD (region STRING)
    return f"ALTER TABLE {table} ADD (region STRING)"


def upsert_sql_v2(src: str, tgt: str) -> str:
    # TODO: also carry the new column.
    return f"INSERT INTO {tgt} SELECT id, name, amt FROM {src}"

Hints

Hint 1

The target table must be upsert-capable: declare a PRIMARY KEY (id) NOT ENFORCED and WITH ('format-version'='2'). Without the PK/v2 the sink can only append.

Hint 2

The upsert is just INSERT INTO <tgt> SELECT <cols> FROM <src>; the debezium-json source already carries the +I/-U/+U/-D RowKinds, so Flink applies updates/deletes by the primary key.

Hint 3

Schema evolution: ALTER TABLE <tgt> ADD (region STRING). Old rows read NULL for region; the follow-up upsert selects the new column too.

Run it locally

uv run de start FLK-10/01-cdc-to-iceberg
# edit workspace/FLK-10/01-cdc-to-iceberg/
uv run de test