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.2; 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.2).

Quickstart

A daily PostgreSQL → Parquet DAG on one worker, with SQLite state. Run the commands on the worker (or in the image) where Airflow runs.

1. Generate the config from the live database with rivet init (the rivet that rivet-cli installed; 0.32.2 or newer for RIVET_STATE_URL=sqlite:), and keep its state in the directory the DAG will use. The directory must exist, on persistent storage, at the same path on every worker:

   mkdir -p /var/lib/rivet/postgres
   export RIVET_PG_URL='postgresql://user:password@db-host:5432/app'
   RIVET_STATE_URL=sqlite:/var/lib/rivet/postgres/.rivet_state.db \
     rivet init --source-env RIVET_PG_URL --include 'users' 'orders' -o /opt/pipelines/postgres.yaml
   

   Review the file: one export per table, mode, destination. Without RIVET_STATE_URL, init leaves .rivet_state.db beside the config and the operator refuses to start from another one (RIVET_AIRFLOW_STATE_BESIDE_CONFIG).

2. Add the Connection the task builds RIVET_PG_URL from (type postgres; generic is refused):

   airflow connections add rivet_postgres --conn-uri 'postgres://user:password@db-host:5432/app'
   

3. Write the DAG. The builders default to schedule=None (manual runs); pass your own:

   from datetime import datetime
   
   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"],
       load=False,  # no warehouse yet: True adds load tasks, which need a `load:` block
       schedule="@daily",
       start_date=datetime(2026, 10, 1),
       catchup=False,
       operator_kwargs={"env_from_connections": {"RIVET_PG_URL": "rivet_postgres"}},
   )
   

4. Trigger it. Each task logs one line, e.g. rivet apply users → OK: 300 rows, 1 file in 0.3 s (run users_20261020T165708.134_54927). The Parquet lands in the config's destination; a relative path such as init's ./output/users/ is resolved by rivet against the directory of the config it reads, which under the operator is state_dir: /var/lib/rivet/postgres/output/users/. Write an absolute path to put it elsewhere.

If the worker has an older rivet on its PATH, the operator refuses it with both versions (RIVET_AIRFLOW_BINARY_TOO_OLD); remove it or pass rivet_bin=. Several workers, Kubernetes pods or a shared state: see State.

Documentation