Sinks

A sink writes a batch and a checkpoint. An exactly-once sink writes both atomically, so a crash commits both or neither.

Delivery

sink	guarantee	resume from	multi-instance
DeltaSink	exactly-once	checkpoint in the Delta commit metadata	yes
PostgresSink	exactly-once	checkpoint row keyed by pipeline_id	single instance
KafkaSink	exactly-once	checkpoint record on a compacted topic	single instance
DatabricksSink	exactly-once	checkpoint in the Delta commit metadata	yes
S3Sink	at-least-once	source’s committed position	idempotent (deterministic keys)
SnowflakeSink	exactly-once	checkpoint row keyed by pipeline_id	single instance
BigQuerySink	at-least-once	source’s committed position	idempotent (key-based MERGE)

An at-least-once sink writes data only; the source’s committed position is the resume authority, and idempotent writes keep replays from duplicating.

Delta

from pyrula.pipelines import DeltaSink

DeltaSink("/data/orders", pipeline_id="orders-to-delta")

Writes the Arrow batch and the checkpoint in one Delta transaction, the checkpoint in the commit metadata keyed by pipeline_id. Resume scans history for the latest checkpoint that pipeline wrote. Under a multi-instance rebalance it merges per-partition offsets to the max, so two instances writing the same table do not under-resume each other’s partitions.

Postgres

from pyrula.pipelines import PostgresSink

PostgresSink(dsn, table="orders_sink", pipeline_id="orders-to-pg")

Ingests the batch with ADBC adbc_ingest (COPY under the hood) and upserts the checkpoint into a __pyrula_checkpoints table, both in one Postgres transaction. The target table is created from the Arrow schema on first write.

Kafka

from pyrula.pipelines import KafkaSink

KafkaSink(bootstrap_servers, topic="orders-mirror", pipeline_id="mirror")

Produces the data records and a checkpoint record in one Kafka transaction, the checkpoint on a compacted topic keyed by pipeline_id. A read_committed consumer never sees an aborted batch. Resume reads the last checkpoint record for the pipeline.

v1 is raw-only: it produces the batch’s binary value column as is. Encode upstream and feed value_format="raw".

Databricks (Unity Catalog)

from pyrula.pipelines import DatabricksSink

DatabricksSink("main.sales.events", pipeline_id="orders-to-uc")

A DeltaSink that resolves the UC table and vends temporary credentials first, then writes the object store directly. Exactly-once is inherited from the Delta sink unchanged: data and checkpoint land in one Delta commit, keyed by pipeline_id. Credentials are re-vended before they expire.

Auth and constraints match the Databricks source: ambient SDK config or host=/token=/ client=, AWS only, external-storage UC tables, pyrula-pipelines[databricks]. The table must already exist (v1 appends, it does not create UC-managed tables).

S3 (object store)

from pyrula.pipelines import S3Sink

S3Sink("bucket/prefix", pipeline_id="orders-to-s3", storage_options={"AWS_REGION": "us-east-1"})

Writes each batch as one Parquet object under bucket/prefix/<pipeline_id>/, named deterministically from the checkpoint. A replay writes the same key and overwrites rather than duplicating, so the result is effectively-once even though the sink itself is at-least-once. storage_options takes the same AWS_* keys as the Delta connector and reaches any S3-compatible store; pass AWS_ENDPOINT_URL for a self-hosted backend (for example Garage). Parquet only, snappy-compressed. Give the prefix as bucket/prefix, with no s3:// scheme. pyrula-pipelines[s3].

Snowflake

from pyrula.pipelines import SnowflakeSink

SnowflakeSink(dsn, table="orders_sink", pipeline_id="orders-to-sf")

Ingests the batch with ADBC and merges the checkpoint into a __pyrula_checkpoints table, both in one transaction, so data and checkpoint commit together. The target table is created from the Arrow schema on first write. pyrula-pipelines[snowflake].

BigQuery

from pyrula.pipelines import BigQuerySink

BigQuerySink(dsn, table="orders_sink", key=["id"], pipeline_id="orders-to-bq")

At-least-once. BigQuery has no transaction spanning the data write and a checkpoint, so idempotency comes from a key-based merge: each batch is staged and merged into the target on key, so a replay of the same rows is absorbed instead of duplicated. key is required and names the columns that identify a row. pyrula-pipelines[bigquery].