Apache Iceberg¶

IcebergIOHandler¶

Persists asset outputs as Apache Iceberg tables, through a pyiceberg catalog. See the Iceberg guide.

from rivers.io_handlers.iceberg import IcebergIOHandler

io = IcebergIOHandler(
    catalog_name="lake",
    catalog_properties={"type": "rest", "uri": "https://catalog.example.com"},
    namespace="analytics",
)

Constructor:

Asset metadata overrides:

Output metadata:

Methods for action bodies¶

An action gets the asset's resolved handler as ctx.io_handler. These methods use the same table and partition resolution as the writes.

pyiceberg_catalog()¶

Loads the pyiceberg catalog: load_catalog(catalog_name, **catalog_properties).

table_identifier(asset_name, asset_metadata=None)¶

Returns (*namespace, table), honoring iceberg/namespace and iceberg/table.

load_table(asset_name, asset_metadata=None)¶

Returns the asset's pyiceberg Table. Raises NoSuchTableError when the table does not exist yet.

partition_filter(asset_metadata, partition, schema)¶

Returns the pyiceberg expression for the rows of the partition(s), honoring iceberg/partition_expr. Pass ctx.asset_metadata, ctx.partition, and table.schema(); the schema types the time-window bounds.

@rs.action(outcome=rs.Outcome.Unmaterialize)
@classmethod
def delete(cls, ctx: rs.ActionContext) -> None:
    table = ctx.io_handler.load_table(ctx.asset_name, ctx.asset_metadata)
    table.delete(ctx.io_handler.partition_filter(ctx.asset_metadata, ctx.partition, table.schema()))

partition is required: a keyless action run raises ValueError.

resolved_snapshot_properties(asset_metadata)¶

Returns snapshot_properties with the asset's iceberg/snapshot_properties on top.

UpsertConfig¶

How mode="upsert" matches rows, with pyiceberg's Table.upsert.

from rivers.io_handlers.iceberg import UpsertConfig

config = UpsertConfig(join_cols=["id"])

IcebergAsset¶

Asset base class with Iceberg maintenance verbs. Subclass it and define materialize; delete and expire_snapshots appear as actions, run against the asset's IcebergIOHandler (its own io_handler or the repository default). Sets kinds = "iceberg".

from rivers.io_handlers.iceberg import IcebergAsset  # also exported as rs.IcebergAsset

class Orders(IcebergAsset):
    io_handler = LAKE

    @classmethod
    def materialize(cls) -> pl.DataFrame: ...

A verb needs the asset to resolve an IcebergIOHandler; another handler fails the step with a TypeError. On a table that does not exist yet, expire_snapshots reports ActionResult.unchanged(), and delete still applies its Unmaterialize outcome. A subclass that defines a verb replaces the built-in one.

ExpireSnapshotsConfig¶

Per-asset overrides via run_action("expire_snapshots", config={"assets": {"<asset>": {"config": {...}}}}):

IcebergTypeHandler¶

Abstract base for adding type support to IcebergIOHandler.

from rivers.io_handlers.iceberg import IcebergReadRequest, IcebergTypeHandler, IcebergWriteRequest

class MyTypeHandler(IcebergTypeHandler[MyType]):
    @property
    def supported_types(self) -> Sequence[type[MyType]]:
        return [MyType]

    def load_input(self, request: IcebergReadRequest, target_type: type[MyType]) -> MyType:
        ...

    def handle_output(self, context: OutputContext, obj: MyType, request: IcebergWriteRequest) -> None:
        ...

Abstract members:

IcebergReadRequest has the catalog, the table identifier, the partition and partition_expr, the columns, and the snapshot_id. Its helpers are table (the pyiceberg table, loaded on first use), scan() (a pyiceberg scan with the partition filter, the columns, and the snapshot), and predicate (the partition as a SQL WHERE clause). IcebergWriteRequest has the resolved write settings.

ArrowIcebergTypeHandler¶

IcebergTypeHandler for types that convert to Arrow. Writes with handle_output are implemented with pyiceberg; subclasses implement to_arrow and load_input.

from rivers.io_handlers.iceberg import ArrowIcebergTypeHandler

class MyTypeHandler(ArrowIcebergTypeHandler[MyType]):
    @property
    def supported_types(self) -> Sequence[type[MyType]]:
        return [MyType]

    def to_arrow(self, obj: MyType) -> pa.Table | pa.RecordBatchReader:
        ...

    def load_input(self, request: IcebergReadRequest, target_type: type[MyType]) -> MyType:
        return convert(request.scan().to_arrow())

Built-in handlers (registered when their library is installed):

PySparkIcebergTypeHandler¶

IcebergTypeHandler-based handler class for adding Spark-based type support to IcebergIOHandler. Reads with load_input and writes with handle_output, both are implemented with Spark.

from rivers.io_handlers.iceberg import PySparkIcebergTypeHandler

class MyTypeHandler(PySparkIcebergTypeHandler[MyType]):
    @property
    def supported_types(self) -> Sequence[type[MyType]]:
        return [MyType]

    def load_input(
        self, request: IcebergReadRequest, target_type: type[MyType]
    ) -> MyType:
        """Load a lazy Spark DataFrame with projection, partition, and snapshot filters."""
        ...

    def handle_output(
        self,
        context: OutputContext,
        obj: DataFrame,
        request: IcebergWriteRequest,
    ) -> None:
        """Write through Spark and attach Rivers output metadata."""
        ...