Rivet

Declarative SQL pipelines with multi-engine execution, quality checks, and built-in testing.

Rivet is a framework that revolutionizes data pipelines by strictly separating concerns. It allows you to define your pipeline once and run it on DuckDB, Polars, PySpark, Postgres or any other engine without changing your logic.

Rivet pipelines are built on three foundational pillars:

This architecture lets you build portable pipelines. Adjacent SQL joints assigned to the same engine are automatically fused into a single query to reduce memory pressure and avoid unnecessary data movement.

• 🔄 Multi-Engine Execution: Swap compute engines without rewriting pipelines.
• 🛠️ Declarative Flexibility: Define joints using SQL, YAML, or Python.
• 🛡️ Ironclad Data Quality: * Assertions run pre-write on computed data to catch errors before they hit your target.
  • Audits run post-write by reading back from the target catalog to verify state.
• 🧪 Built-in Offline Testing: Validate your transformation logic using offline fixture data without needing a live database.
• 💻 Interactive REPL: Use rivet repl for a full-screen terminal UI to explore data, run ad-hoc queries, and iterate on pipeline logic.
• 🔀 Advanced Write Strategies: Supports 7 write modes including append, replace, merge, and scd2 (Slowly Changing Dimensions).

Install Rivet with all plugins:

pip install 'rivetsql[all]'

Or install only what you need:

pip install 'rivetsql[duckdb]'    # recommended for local dev

Scaffold a new project with the required directory structure:

rivet init my_pipeline
cd my_pipeline

Compile and execute your DAG:

rivet run

Three files. Source → Transform → Sink. That's it.

1. Read raw data from a catalog:

-- sources/raw_orders.sql
-- rivet:name: raw_orders
-- rivet:type: source
-- rivet:catalog: local
-- rivet:table: raw_orders
select * from raw_orders

2. Transform with plain SQL:

-- joints/daily_revenue.sql
-- rivet:name: daily_revenue
-- rivet:type: sql
SELECT
    order_date,
    SUM(amount) AS revenue
FROM raw_orders
WHERE status = 'completed'
GROUP BY order_date

3. Write results with quality checks:

-- sinks/daily_revenue_out.sql
-- rivet:name: daily_revenue_out
-- rivet:type: sink
-- rivet:upstream: [daily_revenue]
-- rivet:catalog: warehouse
-- rivet:table: daily_revenue
-- rivet:write_strategy: replace
-- rivet:assert: not_null(revenue)
-- rivet:assert: row_count(min=1)

$ rivet run
✓ compiled 3 joints in 38ms
  raw_orders          ✓ OK (1200 rows)
  daily_revenue       ✓ OK (90 rows)
  daily_revenue_out   ✓ OK (90 rows)

  38ms | 3 joints | 1 groups | 0 failures

If an assertion like not_null fails, the write is completely aborted, keeping your target clean.

Rivet is fully extensible through plugins.

Start here:

• Getting Started guide
• Concepts Overview
• Testing Guide
• Quality Checks Guide

Pull requests are welcome! Check out our Contribution Guidelines.

git clone https://github.com/rivetsql/rivetsql