Quickstart

pip install 'pyrula-pipelines[delta]'.

The recipe

kafka_to_delta wires a Kafka source, a Delta sink, and the engine into one pipeline. It builds the pipeline; it does not run it.

from pydantic import BaseModel
from pyrula.kafka import KafkaConsumerConfig
from pyrula.pipelines import kafka_to_delta

class Order(BaseModel):
    id: int
    total: float
    note: str | None = None

cfg = KafkaConsumerConfig(
    bootstrap_servers="localhost:9092",
    topics=["orders"],
    group_id="orders-to-delta",
    auto_offset_reset="earliest",
)

pipe = kafka_to_delta(cfg, "/data/orders", pipeline_id="orders-to-delta",
                      value_format="avro", schema=Order)
pipe.run()

value_format="raw" skips decode and gives you a single binary value column. "avro" and "json" decode in Rust to one column per field. The schema is the Avro schema for both: pass a pydantic model (or dataclass) and it is derived for you, the same model you would give @kafka_agent, or pass an Avro schema JSON string. An unknown value_format or a missing schema is caught when you build the pipeline, not mid-stream.

Run modes

run() is service mode. It polls until you call stop() or the process gets SIGTERM. run(max_batches=N) stops after N batches or when the source drains, which is what you want for a bounded backfill.

By hand

The recipe is three objects. Assemble them yourself when you want a different source or sink.

from pyrula.pipelines import Pipeline, KafkaSource, DeltaSink

pipe = Pipeline(
    source=KafkaSource(cfg, value_format="avro", schema=Order),
    sink=DeltaSink("/data/orders", pipeline_id="orders-to-delta"),
    transform=None,        # optional callable on each Arrow batch
    max_retries=3,         # sink write retries with backoff before halting
)

A transform is a function from an Arrow batch to an Arrow batch, run between poll and write. Leave it out for a straight copy. For Polars expressions and read pushdown, see Transforms and pushdown.

The decorator

When the transform is the point of the pipeline, @pipeline reads better than passing a callable. The function body is the transform; the source, the sink, and any engine options go on the decorator. The decorated name is a ready-to-run Pipeline.

import pyarrow as pa
import pyarrow.compute as pc
from pyrula.pipelines import pipeline, KafkaSource, DeltaSink

@pipeline(
    source=KafkaSource(cfg, value_format="avro", schema=Order),
    sink=DeltaSink("/data/orders", pipeline_id="orders-to-delta"),
    max_retries=3,
)
def big_orders(batch):
    t = pa.table(batch)
    return t.filter(pc.greater(t.column("total"), 100))

big_orders.run()

The batch is Arrow in and Arrow out, the same contract as a plain transform. The raw function is still reachable as big_orders.transform for a unit test that does not touch the engine. Keyword arguments after source and sink pass straight to Pipeline. A view argument is reserved for a future zero-copy Polars view; today only view="arrow" (the default) is supported.

Resume

Stop the pipeline, start a new one with the same pipeline_id against the same table. The Delta sink reads the last checkpoint it committed and the Kafka source seeks past it. No duplicates, no gaps. A crash takes the same path on the next start.