Delta

pyrula.connectors.delta reads and writes Delta Lake tables. Work with @case models or with Arrow directly. Every call returns an Either, so errors surface as values instead of exceptions.

pip install 'pyrula-connectors[delta]'

DeltaTable connects to a table via a path string and an optional storage_options dict. The storage_options dict passes through to the underlying delta-rs library, so you can reach any object store it supports (S3, Azure Blob, GCS) by supplying the appropriate credentials.

Models

from pyrula import case, IList
from pyrula.connectors.delta import DeltaTable

@case
class Order:
    id: int
    item: str

# create returns Either[DeltaTable]; unwrap with .value or handle the error
table = DeltaTable.create("/data/orders", Order).value

table.append(IList([Order(1, "a"), Order(2, "b")]))  # Ok(WriteReport(...))
rows = table.read(Order).value                        # IList[Order]

DeltaTable.create takes the model class and optional partition_by, storage_options, and table_properties (for example to turn on Change Data Feed). To open an existing table, construct DeltaTable(path) directly.

append, overwrite, merge, delete, and update cover writes. Each returns Either[WriteReport] or, for merge, Either[MergeReport]. read loads the current snapshot; pass version= or timestamp= to load a past one. An optional columns= list and a filters= expression (PyArrow format) work on both.

For replay-safe writes, prefer merge_keyed(data, keys=["id"]) over append: it upserts on the key set, so a workflow step that re-runs on retry or recovery cannot double-write. append is only safe where the engine guarantees exactly-once execution.

Arrow

For columnar data, skip the model layer.

import pyarrow as pa
from pyrula.connectors.delta import DeltaTable

# write_arrow defaults to mode="append"; pass mode="overwrite" to replace
DeltaTable.write_arrow("/data/orders", pa.table({"id": [1, 2]}))

t = DeltaTable("/data/orders")
snapshot = t.read_arrow().value                   # pyarrow.Table
for batch in t.scan_batches(batch_size=10_000):   # iterator of RecordBatch
    ...

write_arrow is a static method and accepts any Arrow table or record batch. It returns Either[WriteReport]. read_arrow returns the whole snapshot; scan_batches streams it without loading the full table into memory.

load_cdf(starting_version=...) returns Change Data Feed rows as a pyarrow table (requires delta.enableChangeDataFeed=true). added_files(from_version, to_version) lists the parquet paths added in that commit range, useful as a CDF fallback.

Cloud storage

Pass a storage_options dict at construction or to any static method:

from pyrula.connectors.delta import DeltaTable

s3_opts = {
    "AWS_ACCESS_KEY_ID": "...",
    "AWS_SECRET_ACCESS_KEY": "...",
    "AWS_REGION": "us-east-1",
}

table = DeltaTable("s3://my-bucket/orders", storage_options=s3_opts)
rows = table.read(Order).value

Azure Blob Storage, GCS, and other backends supported by delta-rs work the same way. Key names and credential fields follow the delta-rs storage documentation.

Maintenance

table.optimize()          # compact small files
table.vacuum(168)         # remove files older than 168 hours
table.history(limit=5)    # Ok(IList[dict])

vacuum and optimize both return Either.

Pipelines

The Pipelines connectors build on this. DeltaSink writes batches with an exactly-once checkpoint in the commit metadata. DeltaSource reads a table incrementally by version (local paths only in v1).