Data connectivity & integrationIceberg tablesIncremental and CDCIncremental code examples

Incremental code examples

This page provides code examples for incremental Iceberg transforms. For more background on the Foundry APIs used below, see Incremental Python transforms with Iceberg and Iceberg changelogs and CDC pipelines. For a conceptual introduction to working with Iceberg tables incrementally, see Incremental processing with Iceberg tables.

Append-only incremental using Foundry APIs

In an append-only incremental transform, both the read and the write Foundry APIs default to their incremental behavior: reads return the rows appended since the last build, and writes append to the output rather than replacing it.

The table_read_mode="append_only" option is included here for clarity. You can omit it and get the same behavior because append-only is the default read mode.

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.polars()
    active = added_rows.filter(pl.col("status") == "active")
    output.write_table(active)

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: LightweightContext, source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    conn = ctx.duckdb().conn
    conn.register("added_rows", source.arrow())
    query = conn.sql("SELECT * FROM added_rows WHERE status = 'active'")
    output.write_table(query.to_arrow_table())

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.pandas()
    active = added_rows[added_rows["status"] == "active"].astype({"status": "string"})
    output.write_table(active)

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


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalTableTransformInput, output: TableTransformOutput):
    added_rows = source.dataframe()
    active = added_rows.filter(added_rows.status == "active")
    output.write_dataframe(active)

Append-only incremental using native Iceberg write APIs

These examples use Foundry APIs to read appended rows and native open source Iceberg APIs to append updates.

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
from transforms.api import incremental, transform
from transforms.tables import (
    IncrementalIcebergInput,
    IncrementalIcebergOutput,
    TableInput,
    TableOutput,
)
from pyiceberg.expressions import EqualTo


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    
    # Use Foundry APIs to retrieve the added rows & output
    active = source.arrow(row_filter=EqualTo("status", "active"))
    output_table = output.table()

    # Handle first run case
    if output_table is None:
        output.write_table(active)
        return

    # Use PyIceberg append to write incrementally
    with output_table.transaction() as txn:
        txn.append(active)

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
from pyspark.sql.functions import col, days
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):

    # Use Foundry APIs to retrieve the added rows    
    added_rows = source.dataframe()
    active = added_rows.filter(col("status") == "active")

    # Write with Iceberg's DataFrameWriterV2 API, partitioned by a transform of a column
    writer = active.writeTo(output.identifier).partitionedBy(days(col("ts")))

    # Use ctx.is_incremental to differentiate between incremental and non-incremental resolution cases
    if ctx.is_incremental:
        writer.append()
    else:
        writer.createOrReplace()

Changelog incremental using Foundry APIs

This example demonstrates how to read an Iceberg changelog and apply it to an output table based on a unique identifier using Foundry's changelog APIs.

The example assumes id is unique in the source and keeps only active rows in the output. When a row becomes inactive, it sends a DELETE to filter the inactive row from the output.

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
from transforms.api import (
    IncrementalTableTransformInput,
    IncrementalTransformContext,
    incremental,
    transform
)
from transforms.tables import TableInput, TableOutput, TableTransformOutput
from pyspark.sql import functions as F


@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
):
    if not ctx.is_incremental:
        output.write_dataframe(source.dataframe().filter(F.col("status") == "active"))
        return

    changelog_df = source.changelog(["id"])

    reconciled = changelog_df.withColumn(
        "_change_type",
        F.when(
            (F.col("_change_type") != "DELETE")
            & (F.col("status") == "active"),
            F.lit("INSERT"),
        ).otherwise(F.lit("DELETE")),
    )

    output.apply_changelog(reconciled, ["id"])

Changelog incremental using Iceberg APIs

Call Iceberg's create_changelog_view procedure ↗ directly when you need options that Foundry's API does not expose. Without a snapshot range, it reads from the table's first snapshot.

The native procedure includes UPDATE_BEFORE rows and intermediate changes across snapshots. This example stores the last processed snapshot in an output table property and selects the latest action before merging. It assumes id is unique in the source, with one action per key per snapshot after removing UPDATE_BEFORE.

This example shows the basic structure of the merge. Consider the appropriate MERGE INTO logic carefully in the context of your pipeline.

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
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"  # Last source snapshot ID processed


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

    spark = ctx.spark_session
    source_tbl = source.identifier
    output_tbl = output.identifier

    # The snapshot this build should read up to: the source's current snapshot.
    end = spark.sql(f"""
        SELECT snapshot_id FROM {source_tbl}.snapshots
        ORDER BY committed_at DESC LIMIT 1
    """).collect()
    end = end[0]["snapshot_id"] if end else None

    # First build, or a build that could not resolve incrementally (using Foundry APIs)
    if not ctx.is_incremental:
        source_df = source.dataframe()
        output.write_dataframe(source_df.filter(source_df.status == "active"))
        # Set watermark so the first incremental build has a start point
        spark.sql(f"ALTER TABLE {output.identifier} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")
        return

    # The last snapshot previously processed, read back from the output table property.
    start = spark.sql(f"""
        SHOW TBLPROPERTIES {output_tbl} ('{_WATERMARK_PROP}')
    """).collect()[0]["value"]

    if start == str(end):
        # Nothing new to process.
        return

    # Build the changelog over the range (start, end]
    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_tbl}',
            changelog_view => 'source_changelog',
            identifier_columns => array('id'),
            options => map('start-snapshot-id', '{start}', 'end-snapshot-id', '{end}')
        )
    """)

    # Apply the deduplicated changelog
    spark.sql(f"""
        MERGE INTO {output_tbl} AS tgt
        USING (
            SELECT * FROM (
                SELECT *,
                    ROW_NUMBER() OVER (
                        PARTITION BY id ORDER BY _change_ordinal DESC
                    ) AS _rn
                FROM source_changelog
                WHERE _change_type != 'UPDATE_BEFORE'
            )
            WHERE _rn = 1
        ) AS src
        ON src.id = tgt.id
        WHEN MATCHED AND (src._change_type = 'DELETE' OR src.status != 'active') THEN DELETE
        WHEN MATCHED AND src.status = 'active' THEN UPDATE SET *
        WHEN NOT MATCHED AND src._change_type != 'DELETE' AND src.status = 'active' THEN INSERT *
    """)

    # Update the watermark to track latest read snapshot
    spark.sql(f"ALTER TABLE {output_tbl} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")

←

PREVIOUSIncremental Iceberg transforms

NEXTChangelogs and CDC

→