# 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](#construct-an-incremental-transform-with-the-changelog-table-read-mode).
2. [Generate a changelog](#generate-a-changelog-from-your-input-table) from the input table.
3. [Apply the changes](#apply-the-changelog-to-your-output-table) to the output table.
4. [Handle non-incremental builds](#handle-the-non-incremental-build-case) explicitly.

To jump directly to runnable code, see [Quick start](#quick-start).

## Construct an incremental transform with the `changelog` table read mode

Construct an [incremental transform](../incremental-transforms/index.md) as usual, taking care to specify the [table read mode](../incremental-transforms/index.md#incremental-decorator-and-read-mode) as `changelog`.

```python
@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 ↗](https://iceberg.apache.org/docs/latest/spark-procedures/#create_changelog_view).

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:

|Method|Identifier changelog|Net changes changelog|
|---|---|---|
|Foundry `.changelog` API|Pass identifiers, for example `source.changelog(["primary_key"])`. Returns the latest change per unique identifier and removes `UPDATE_BEFORE` rows.|Omit identifiers: `source.changelog()`. Returns net inserts and deletes across the snapshot range.|
|Open-source [`create_changelog_view` procedure ↗](https://iceberg.apache.org/docs/latest/spark-procedures/#create_changelog_view)|Pass `identifier_columns => array('primary_key')`. Returns all changes across the requested snapshots, including `UPDATE_BEFORE` and `UPDATE_AFTER` rows.|Omit identifiers and set `net_changes => true`. Returns net inserts and deletes across the snapshot range.|

:::callout{theme="neutral" title="Identifier changelogs and net changes changelogs"}
An identifier changelog groups changes by the column names you provide as identifiers. A net changes changelog matches rows using all data columns, cancels matching inserts and deletes across the snapshot range, and preserves duplicate row counts. Foundry's identifier changelog API requires unique identifiers; if the source contains duplicate rows for an identifier, changelog creation fails. Use a net changes changelog when you cannot rely on your keys. For examples of both modes, see the [changelog API technical primer](../changelog-primer/index.md#example-changelog-evolution).
:::

## 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` ↗](https://iceberg.apache.org/docs/latest/spark-writes/#merge-into) statement:

|Method|What it does|When to use it|
|---|---|---|
|Foundry `.apply_changelog`|Upserts `INSERT` and `UPDATE_AFTER` rows and deletes `DELETE` rows, matching on the identifiers you provide. Requires at most one action per output key.|Standard identifier-based upserts and deletes.|
|Open-source Spark SQL [`MERGE INTO` statement ↗](https://iceberg.apache.org/docs/latest/spark-writes/#merge-into)|Applies the matching conditions and update, insert, and delete clauses you write.|Custom matching, conditions, or business logic in the merge.|

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

```python
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:

```sql
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 *
```

:::callout{theme="warning" title="Propagate changes to filtered-out rows"}
If your transformation filters rows, ensure that rows that stop matching the filter are removed from the output. Dropping their changelog rows leaves the previous output rows unchanged. <br><br>
With Foundry's `.apply_changelog()`, convert changes for rows that no longer match the filter to `DELETE` before applying them. With open-source Spark SQL, you can instead express the filtering logic in your `MERGE INTO` clauses. See the [filtered-output examples](../incremental-code-examples/index.md#changelog-incremental-using-foundry-apis).
:::

## 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.

```python tab="Foundry APIs"
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())
```

```python tab="Open-source Iceberg"
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-primer/index.md)
* [Changelog incremental code examples with row filters](../incremental-code-examples/index.md#changelog-incremental-using-foundry-apis)