Examples

Short, practical snippets for common pipeline shapes. Each one uses the real API.

Kafka to Delta (service mode)

The kafka_to_delta recipe covers the common case. Pass handle_signals=True so a SIGTERM from your process supervisor stops the loop cleanly.

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

class Event(BaseModel):
    id: int
    user_id: int
    action: str

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

pipe = kafka_to_delta(
    cfg,
    "/data/lake/events",
    pipeline_id="events-to-delta",
    value_format="avro",
    schema=Event,
    handle_signals=True,
)
pipe.run()

Resume is automatic. Restart the process with the same pipeline_id and table_uri and it continues from the last committed checkpoint.

Bounded backfill (Kafka to Delta)

run(max_batches=N) stops after N batches or when the source drains. Use it to replay a fixed window of history without writing a service loop.

pipe = kafka_to_delta(
    cfg,
    "/data/lake/events",
    pipeline_id="backfill-events",
    value_format="avro",
    schema=Event,
)
committed = pipe.run(max_batches=500)
print(f"committed {committed} batches")

Set auto_offset_reset="earliest" on the consumer config to start from the beginning of the topic.

Delta (local) to Postgres

Read incremental commits from a Delta table on local disk and land them in Postgres. Both source and sink are exactly-once, so resume is driven by the sink checkpoint.

from pyrula.pipelines import DeltaSource, Pipeline, PostgresSink

DSN = "postgresql://user:pass@localhost:5432/mydb"

pipe = Pipeline(
    source=DeltaSource("/data/lake/orders", mode="append_only"),
    sink=PostgresSink(DSN, table="orders_mirror", pipeline_id="delta-to-pg"),
)
pipe.run(max_batches=100)

append_only mode reads only the parquet files each new commit added. In v1 the table path must be on local disk; object-store URIs are not yet supported for this mode.

Databricks (Unity Catalog) to Postgres

Move a Unity Catalog table into Postgres with the databricks_to_postgres recipe. It vends temporary credentials and reads the object store directly, no SQL warehouse in the path.

from pyrula.pipelines import databricks_to_postgres

DSN = "postgresql://user:pass@localhost:5432/mydb"

pipe = databricks_to_postgres(
    uc_table="main.sales.events",
    dsn=DSN,
    pg_table="events_mirror",
    pipeline_id="uc-to-pg",
)
pipe.run(max_batches=100)

Auth is the ambient SDK config by default; pass host=/token=/client= to override. postgres_to_databricks is the reverse. AWS only in v1, and the UC table must be on customer-managed external storage. Needs pyrula-pipelines[databricks].

Kafka to Kafka (mirror, raw)

KafkaSink is raw-only in v1: the batch’s value column must be binary. Use value_format="raw" on the source so the bytes pass through without decode.

from pyrula.pipelines import KafkaSource, KafkaSink, Pipeline
from pyrula.kafka import KafkaConsumerConfig

src_cfg = KafkaConsumerConfig(
    bootstrap_servers="source-kafka:9092",
    topics=["payments"],
    group_id="mirror-payments",
    auto_offset_reset="earliest",
)

pipe = Pipeline(
    source=KafkaSource(src_cfg, value_format="raw"),
    sink=KafkaSink(
        "dest-kafka:9092",
        topic="payments-mirror",
        pipeline_id="payments-mirror",
    ),
)
pipe.run()

The checkpoint record lands on a compacted __pyrula_pipeline_checkpoints topic, keyed by pipeline_id. Resume reads that topic at startup.

Custom sink with a transform

Wire a custom sink together with a filter transform. The transform runs between poll and write; it receives and returns an Arrow batch.

import pyarrow as pa
import pyarrow.compute as pc
from pydantic import BaseModel
from pyrula.pipelines import DeliveryGuarantee, KafkaSource, Pipeline, Sink
from pyrula.kafka import KafkaConsumerConfig

class Order(BaseModel):
    id: int
    total: float

class PrintSink(Sink):
    delivery = DeliveryGuarantee.AT_LEAST_ONCE

    def open(self, schema): ...
    def write_batch(self, batch, checkpoint):
        print(pa.table(batch).to_pandas())
    def resume_checkpoint(self):
        return None
    def close(self): ...

cfg = KafkaConsumerConfig(
    bootstrap_servers="kafka:9092",
    topics=["orders"],
    group_id="debug-orders",
    auto_offset_reset="earliest",
)

def keep_large(batch):
    t = pa.table(batch)
    return t.filter(pc.greater(t.column("total"), 100))

pipe = Pipeline(
    source=KafkaSource(cfg, value_format="avro", schema=Order),
    sink=PrintSink(),
    transform=keep_large,
    max_retries=0,
)
pipe.run(max_batches=5)

The same pipeline reads as one unit with the @pipeline decorator, where the function body is the transform:

from pyrula.pipelines import pipeline

@pipeline(
    source=KafkaSource(cfg, value_format="avro", schema=Order),
    sink=PrintSink(),
    max_retries=0,
)
def keep_large(batch):
    t = pa.table(batch)
    return t.filter(pc.greater(t.column("total"), 100))

keep_large.run(max_batches=5)