airflow-provider-rivet

Apache Airflow operators and DAG builders that run the rivet binary on the worker as a subprocess and turn its result into Airflow's: retry or stop, which unit failed, and what to do about it. The contract it implements is ADR-0039.

rivet itself (sources, destinations, warehouses, CDC, configs): https://github.com/panchenkoai/rivet.

How it works

• A task runs the rivet binary on the worker, as a subprocess with a minimal environment; no rivet code runs inside Airflow.
• Plan, then apply. RivetApplyOperator seals a plan (rivet plan) and applies it (rivet apply); a retry replays the same sealed plan, a cleared task plans again. load and compact tasks follow per table.
• rivet's exit becomes Airflow's decision: success, a retry for what may pass by itself, a final stop for a refusal. The code, class and action go to XCom and to a short block in the task log; rivet's raw output stays in files.
• DAG builders make the whole batch DAG (plan, apply, load, compact, in waves) or the CDC DAG (one bounded drain, then load and compact per table) from a config.
• State lives in a declared directory on the worker (state_dir, SQLite) or in PostgreSQL (RIVET_STATE_URL); the operator refuses a missing or mismatched one instead of starting from empty state.
• Credentials come from Airflow Connections (databases, storage, warehouses), resolved inside the task; nothing secret is in argv, XCom or the task log.

Status: alpha. It is versioned independently of the binary and checks the binary's version at run time (minimum rivet 0.32.0; a lower one is refused with both versions in the message). Supported: Airflow 2.10+ and 3.x, Python 3.12+ (Airflow 2.10 only on 3.12).

Install

pip install airflow-provider-rivet            # or airflow-provider-rivet[slack]

Install it into the Airflow environment with Airflow's own constraints file, so it does not move Airflow's pins. It depends on rivet-cli, which installs the rivet binary (wheels for Linux glibc 2.28+ x86_64/aarch64 and macOS), so the official Airflow images need nothing else:

FROM apache/airflow:3.3.2
RUN pip install --no-cache-dir airflow-provider-rivet

The operators run rivet from the worker's PATH, else the binary rivet-cli installed into Airflow's environment. To run another build, pass rivet_bin="/abs/path/rivet" (for example one copied from ghcr.io/panchenkoai/rivet:v0.32.0).

Example

from airflow_provider_rivet.dags import build_batch_dag

dag = build_batch_dag(
    "rivet_postgres",
    config="/opt/pipelines/postgres.yaml",
    state_dir="/var/lib/rivet/postgres",
    exports=["users", "orders"],
    operator_kwargs={"env_from_connections": {"RIVET_PG_URL": "rivet_postgres"}},
)

Documentation