Skip to content
Pyrula

Pipelines

pyrula.pipelines moves data between systems with exactly-once delivery. A pipeline is a source, a sink, and a loop between them. The loop polls the source for a batch, hands it to the sink, and the sink commits the data and a resume checkpoint together so a crash never leaves them out of sync.

The batch is Arrow. For Kafka, the decode from wire bytes to an Arrow RecordBatch happens in Rust, so the records never become per-record Python objects on the way to the sink.

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


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


pipe = kafka_to_delta(
    KafkaConsumerConfig(bootstrap_servers="localhost:9092", topics=["orders"],
                        group_id="orders-to-delta", auto_offset_reset="earliest"),
    "s3://lake/orders",
    pipeline_id="orders-to-delta",
    value_format="avro",
    schema=Order,
)
pipe.run()

The model

Section titled “The model”
source.poll() -> (batch, checkpoint) -> transform? -> sink.write_batch(batch, checkpoint)

Checkpoints are opaque to the engine. A source defines its own (Kafka uses per-partition offsets, Delta a table version, Postgres a watermark) and only that source knows how to resume from one. The engine passes the checkpoint through and never reads inside it.

Exactly-once

Section titled “Exactly-once”

An exactly-once sink writes the data and the checkpoint in one atomic operation. DeltaSink puts both in a single Delta transaction, with the checkpoint in the commit metadata. PostgresSink does the COPY and a checkpoint upsert in one Postgres transaction. KafkaSink produces the data and a checkpoint record in one Kafka transaction. On restart the sink reads its last committed checkpoint and the source seeks past it.

The Kafka group commit is advisory for these sinks. It feeds lag dashboards, but the resume authority is the sink’s checkpoint, so a crash between the sink commit and the group commit costs nothing.

Install

Section titled “Install”

The base install is Kafka only. Sinks come as extras.

extrafor
pyrula-pipelines[delta]DeltaSink, DeltaSource
pyrula-pipelines[postgres]PostgresSink, PostgresSource
pyrula-pipelines[databricks]DatabricksSink, DatabricksSource (Unity Catalog)
pyrula-pipelines[s3]S3Sink
pyrula-pipelines[snowflake]SnowflakeSink, SnowflakeSource
pyrula-pipelines[bigquery]BigQuerySink, BigQuerySource
pyrula-pipelines[polars]polars_transform for the middle stage

Next

Section titled “Next”