Advanced usage of underlying Iceberg APIs

Most transforms can be expressed using the standard dataframe syntax. However, there are some Iceberg operations which cannot be expressed through the standard dataframe APIs or may benefit otherwise from direct use of the more specific and expressive underlying Iceberg APIs.

For example:

• Row-level operations: UPDATE, DELETE, and MERGE INTO statements that edit rows in place rather than rewriting the whole table.
• Iceberg Spark procedures such as create_changelog_view.
• Iceberg metadata inspection: Reading a table's snapshots, schema, and properties.

For these, you can access the open-source Iceberg APIs directly through handles on the runtime object:

In each case the handle plugs into the open-source API where that API expects a table: as a table name in Iceberg Spark SQL (.identifier), or as the Table object in PyIceberg (.table()). Wherever the open-source Iceberg documentation refers to a table name or table object, pass the native handle in its place.

Row edits in PySpark

This code example demonstrates how to edit rows in place without rewriting the whole table using the Iceberg Spark 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
39
from transforms.api import transform, TransformContext
from transforms.tables import TableInput, TableOutput, TableTransformInput, TableTransformOutput


@transform.spark.using(
    updates=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(
    ctx: TransformContext,
    updates: TableTransformInput,
    output: TableTransformOutput,
):
    # Spark identifier (use in place of a table name)
    output_id = output.identifier

    spark = ctx.spark_session
    if spark.catalog.tableExists(output_id):

        updates_id = updates.identifier

        # Upsert with MERGE INTO, using an input registered as a temporary view
        spark.sql(f"""
            MERGE INTO {output_id} AS t
            USING {updates_id} AS s
            ON t.id = s.id
            WHEN MATCHED THEN UPDATE SET *
            WHEN NOT MATCHED THEN INSERT *
        """)

        # Delete rows by predicate
        spark.sql(f"DELETE FROM {output_id} WHERE as_of < '2026-04-01'")

        # Update rows in place
        spark.sql(f"UPDATE {output_id} SET status = 'Archived' WHERE status = 'Stale'")
 
    else:
        # First build: seed the table from the input
        output.write_dataframe(updates.dataframe())

Partitioned incremental writes in PySpark

This code example demonstrates how to write to a table using Iceberg Spark's DataFrameWriterV2 API ↗. The example includes Palantir incremental semantics and use of Iceberg's partition transforms ↗.

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pyspark.sql.functions import days, col
from transforms.api import Input, transform, incremental, IncrementalTransformContext, IncrementalTransformInput
from transforms.tables import TableOutput, TableTransformOutput


@incremental(v2_semantics=True)
@transform(
    input=Input("/path/input"),
    output=TableOutput("/path/input"),
)
def partition_by_day(
    ctx: IncrementalTransformContext,
    input: IncrementalTransformInput,
    output: TableTransformOutput,
):
    writer = input.dataframe().writeTo(output.identifier).\
        partitionedBy(days(col("created_at")))
    if ctx.is_incremental:
        writer.append()
    else:
        writer.createOrReplace()

Row edits in PyIceberg

This code example demonstrates how to edit rows in place without rewriting the whole table using the PyIceberg APIs ↗. Access the native PyIceberg Table object with .table(), then use its row-level methods: upsert() (to mirror MERGE INTO in Iceberg Spark SQL), delete() for predicate-based deletion, and overwrite() to replace rows.

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
from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput
import pyarrow as pa


@transform.using(
    updates=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(updates: IcebergInput, output: IcebergOutput):

    # Read input into memory
    updates_arrow = updates.arrow()

    # Native PyIceberg table
    table = output.table()

    if table is None:
        output.write_table(updates_arrow)
        return

    # Upsert with upsert(), matching on the join column(s)
    table.upsert(updates_arrow, join_cols=["id"])

    # Delete rows by predicate
    table.delete(delete_filter="as_of < '2026-04-01'")

    # Replace rows matching a filter with new rows
    stale = table.scan(row_filter="status = 'Stale'").to_arrow()
    archived = stale.set_column(
        stale.schema.get_field_index("status"),
        "status",
        pa.array(["Archived"] * stale.num_rows, type=pa.string()),
    )
    table.overwrite(archived, overwrite_filter="status = 'Stale'")

Comparing snapshots with PyIceberg

You can access finer-grained PyIceberg APIs ↗ such as metadata inspection by using .table() on your IcebergInput to return the underlying PyIceberg table. This code example demonstrates inspecting individual snapshots in PyIceberg to manually identify appended rows.

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
from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source_table=TableInput("/path/input"),
    output_table=TableOutput("/path/output"),
)
def compute(source_table: IcebergInput, output_table: IcebergOutput):

    # grab snapshot ids
    current = source_table.current_snapshot_id
    previous = source_table.previous_snapshot_id  
    
    # raw PyIceberg table
    iceberg_table = source_table.table()  

    current_rows = iceberg_table.scan(snapshot_id=current).to_polars()

    if previous is None:
        output_table.write_table(current_rows)
        return

    previous_rows = iceberg_table.scan(snapshot_id=previous).to_polars()
    new_rows = current_rows.join(previous_rows, on="id", how="anti")
    output_table.write_table(new_rows)