Iceberg changelogs and CDC pipelines

Use Iceberg changelogs to build change data capture (CDC) pipelines in PySpark transforms. Changelogs let you apply inserts, updates, and deletes from an input table to an output table without reprocessing the full input on every build.

To build a CDC pipeline:

1. Construct an incremental transform with the changelog table read mode.
2. Generate a changelog from the input table.
3. Apply the changes to the output table.
4. Handle non-incremental builds explicitly.

To jump directly to runnable code, see Quick start.

Construct an incremental transform with the changelog table read mode

Construct an incremental transform as usual, taking care to specify the table read mode as changelog.

Copied!
1
@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")

The changelog table read mode indicates that your transform should be processed incrementally where possible, and that it can process inserts, updates, and deletes incrementally.

Generate a changelog from your input table

Generate a changelog from your input table to produce a DataFrame containing the inserts, updates, and deletes between the previously read version of the input data and the current version.

You can generate a changelog with Foundry's source.changelog() API or the open-source create_changelog_view Spark procedure ↗.

Foundry's .changelog() API wraps the open-source procedure and provides two conveniences:

• Automatic snapshot range detection. Foundry infers the starting and ending snapshots from the transform context. With the open-source procedure, you supply the range yourself.
• Latest change selection for identifier changelogs. Foundry removes UPDATE_BEFORE rows and selects the latest change per unique identifier. The open-source procedure includes intermediate changes across snapshots.

Consider using the Foundry API for ease of use. Use the open-source procedure for options that the Foundry API does not expose, or to avoid the performance impact of identifier validation.

Both the Foundry API and the open-source procedure support identifier changelogs and net changes:

Apply the changelog to your output table

After generating a changelog, apply it to the output using Foundry's .apply_changelog() API or an open-source Spark SQL MERGE INTO ↗ statement:

With Foundry's API, pass the changelog DataFrame and the column names used to match it to the output:

Copied!
1
output.apply_changelog(changelog_df, ["primary_key"])

With open-source Spark SQL, express those actions explicitly. For example, given a prepared latest_changes view with one action per key:

Copied!
1
2
3
4
5
6
MERGE INTO output_table AS tgt
USING latest_changes AS src
ON src.primary_key = tgt.primary_key
WHEN MATCHED AND src._change_type = 'DELETE' THEN DELETE
WHEN MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET *
WHEN NOT MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN INSERT *

Handle the non-incremental build case

On the first build, or when the transform cannot resolve incrementally, replace the output with the full input. Use ctx.is_incremental to distinguish this case from an incremental build.

Avoid generating and applying a changelog over the full input history on a non-incremental build. Correlating row changes across a large snapshot range can be less efficient than reading the current input once.

Both quick start examples below use output.write_dataframe(source.dataframe()) for non-incremental builds. If your transform includes business logic, apply it to the full input before writing.

Quick start

This example shows a simple identity transform that replicates changes from a source table to a target table. It assumes the source has one row per key.

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import IncrementalTransformContext, incremental, transform
from transforms.tables import TableInput, TableOutput

@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def cdc_transform(ctx: IncrementalTransformContext, source, output):
    if ctx.is_incremental:
        changelog_df = source.changelog(["primary_key"])
        output.apply_changelog(changelog_df, ["primary_key"])
    else:
        output.write_dataframe(source.dataframe())

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from transforms.api import (
    IncrementalTableTransformInput,
    IncrementalTransformContext,
    incremental,
    transform,
)
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def cdc_transform(
    ctx: IncrementalTransformContext,
    source: IncrementalTableTransformInput,
    output: TableTransformOutput,
):
    spark = ctx.spark_session
    source_tbl = source.identifier
    output_tbl = output.identifier

    # Read the source snapshot to process through in this build.
    snapshots = spark.sql(f"""
        SELECT snapshot_id FROM {source_tbl}.snapshots
        ORDER BY committed_at DESC LIMIT 1
    """).collect()
    end = snapshots[0]["snapshot_id"] if snapshots else None

    # On a non-incremental build, copy the full source and initialize the watermark.
    if not ctx.is_incremental:
        output.write_dataframe(source.dataframe())
        spark.sql(f"""
            ALTER TABLE {output_tbl}
            SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')
        """)
        return

    # Read the last source snapshot processed by the previous build.
    start = spark.sql(f"""
        SHOW TBLPROPERTIES {output_tbl} ('{_WATERMARK_PROP}')
    """).collect()[0]["value"]

    if start == str(end):
        return

    # Create changelog. The start snapshot is excluded; the end snapshot is included.
    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_tbl}',
            changelog_view => 'source_changelog',
            identifier_columns => array('primary_key'),
            options => map('start-snapshot-id', '{start}', 'end-snapshot-id', '{end}')
        )
    """)

    # Keep the latest action per key, excluding the before-image of an update.
    spark.sql(f"""
        MERGE INTO {output_tbl} AS tgt
        USING (
            SELECT * FROM (
                SELECT *,
                    ROW_NUMBER() OVER (
                        PARTITION BY primary_key ORDER BY _change_ordinal DESC
                    ) AS _rn
                FROM source_changelog
                WHERE _change_type != 'UPDATE_BEFORE'
            )
            WHERE _rn = 1
        ) AS src
        ON src.primary_key = tgt.primary_key
        WHEN MATCHED AND src._change_type = 'DELETE' THEN DELETE
        WHEN MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET *
        WHEN NOT MATCHED AND src._change_type IN ('INSERT', 'UPDATE_AFTER') THEN INSERT *
    """)

    # Advance the watermark after applying the changes.
    spark.sql(f"""
        ALTER TABLE {output_tbl}
        SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')
    """)

For more detail and examples, see:

• Changelog API technical primer
• Changelog incremental code examples with row filters