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.

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


@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
):
    # Read only the changes since the last run
    changelog_df = source.changelog(["id"])

    # Keep only the latest change per identifier
    window = Window.partitionBy("id").orderBy(F.col("_change_ordinal").desc())
    latest = (
        changelog_df
        .filter(F.col("_change_type") != "UPDATE_BEFORE")
        .withColumn("_row_number", F.row_number().over(window))
        .filter(F.col("_row_number") == 1)
        .drop("_row_number")
    )

    # INSERT acts as an upsert
    reconciled = latest.withColumn(
        "_change_type",
        F.when(
            (F.col("_change_type") != "DELETE")
            & (F.col("status") == "active"),
            F.lit("INSERT"),
        ).otherwise(F.lit("DELETE")),
    )

    # Merge the reconciled changelog into the output table
    output.apply_changelog(reconciled, ["id"])

Changelog incremental using Iceberg APIs

Foundry's changelog API wraps Iceberg's create_changelog_view Spark procedure ↗, which you can call directly using the table's catalog and identifier when you need procedure options the wrapper does not expose. Note that the procedure reads from the table's first snapshot unless you pass a snapshot range in its options argument. You can do this by manually managing a marker in your output table that indicates what snapshot has been processed so far.

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 overview

→