• Documentation
    • Search documentation
    karat

    +

    K

    API Reference ↗
    Data connectivity & integrationIceberg tablesIncremental and CDCIncremental Iceberg transforms

    Incremental Python transforms with Iceberg

    This page provides a syntactical introduction to working with incremental Python transforms on Iceberg tables. See incremental processing with Iceberg tables for a conceptual overview on working with Iceberg tables incrementally.

    When constructing incremental transforms with Iceberg tables, there are three key areas to keep in mind:

    1. Add the incremental decorator and read mode
    2. Use the incremental runtime objects
    3. Use incremental compute logic

    Annotated code example showing the incremental decorator and read mode, incremental runtime objects, and incremental compute logic.

    Incremental decorator and read mode

    You can set your transform to run incrementally by including the incremental decorator directly above your transform definition.

    The following standard @incremental arguments are relevant when working with Iceberg tables:

    ArgumentDescription
    v2_semanticsMust be True. Iceberg table inputs and outputs are only supported by v2 incremental semantics.
    require_incrementalDefault is False. If True, the transform fails rather than falling back to a full non-incremental read when it cannot resolve incrementally.
    semantic_versionIncrease this number whenever you change the transform's logic in a way that invalidates the existing output. Changing it forces the next run to be non-incremental.
    snapshot_inputsNames of inputs for which a full rewrite does not invalidate the output, such as reference tables. These inputs are always read in full.

    Additionally, you can chain .with_table_incremental_options(table_read_mode="...") onto the decorator to control which kinds of upstream changes the transform can consume incrementally:

    table_read_modeBehavior
    append_only
    (default)    		The transform runs incrementally over append and replace (compaction) snapshots in the read range. If an overwrite or delete snapshot is present in the read range, the transform cannot resolve incrementally and will either build non-incrementally or fail based on your require_incremental setting.
    changelogThe transform runs incrementally over append, replace, overwrite, and delete snapshots in the read range. Changelog read mode is required to call the .changelog() API. Changelog mode is only available for Spark transforms.

    with_table_incremental_options requires transforms-tables version 0.1390.0 or later.

    Incremental runtime objects

    Foundry provides the following runtime objects for reading and writing Iceberg tables incrementally in a transform:

    • Single-node API: IncrementalIcebergInput, IncrementalIcebergOutput
    • PySpark API: IncrementalTableTransformInput, TableTransformOutput

    Incremental compute logic

    On an incremental run, incremental input runtime objects provide the "added" rows by default. To read something else, call a different method such as .changelog().

    Copied!
    1
added_rows = source.polars()
    Copied!
    1
added_rows = source.arrow()
    Copied!
    1
added_rows = source.pandas()
    Copied!
    1
2
added_rows = source.dataframe()
changelog = source.changelog(["id"])

    You can use these runtime objects to write your incremental computation logic as desired, whether you are writing the added rows, applying a changelog, or using the native Iceberg operations to perform upserts and deletes. See incremental code examples for a variety of incremental transforms.