Aqueduct

Self-healing data pipelines. Declarative. Observable. Autonomous.
One blueprint. Spark or DuckDB. Your data never leaves your servers.

Why Aqueduct

A pipeline fails at 3 a.m. over a column rename upstream. Somebody gets paged. They scroll a four-kilobyte stack trace to find a one-line fix. Aqueduct turns that night into a Git-diffable patch, waiting for review in the morning.

Wake up to a pending patch instead of a wall of errors.

• Declarative, not DAG code. Pipelines are YAML Blueprints. No PySpark boilerplate, no operator classes. Bring your own scheduler; Aqueduct is the control plane on top of the engine.
• Self-healing rather than alerting. On failure, an LLM agent diagnoses the root cause and emits a structured patch. The patch must clear guardrail, lineage, and sandbox gates before it touches your pipeline. The agent cannot generate code or touch a shell. A failure it has solved before heals from memory, with zero LLM tokens.
• Observable by construction. Every run, heal attempt, and column-lineage edge lands in a queryable store. Nothing is added to the hot path to make that happen.
• Model-agnostic and local-first. Anthropic natively, or any OpenAI-compatible endpoint: OpenRouter, DeepSeek, Groq, a local 7B on Ollama or LM Studio. Set provider and base_url, done. Point it at a model inside your perimeter and your data, schemas, and error traces never leave your servers. Multi-model cascades escalate to a bigger model only when the small one gets stuck.
• The harness owns correctness; the model only proposes. The LLM does exactly one thing deterministic code cannot: turn an unstructured failure into a structured hypothesis. Everything after that is deterministic. A constrained patch grammar with 14 operations and no code generation. Validation gates. Budget caps. A signature cache. That division of labor is why a small local model holds up here when raw code-generating assistants don't.

Table of contents

• Supported engines
• What you get
• Core concepts
• The healing flow
• Architecture
• Observability dashboard
• Getting started
• How it compares
• References
• Contributing

Supported engines

Apache Spark Distributed batch at cluster scale. Delta Lake, JDBC, custom data sources, the full Blueprint grammar.

DuckDB Single-node production, zero JVM. Ships with the base install. Same blueprint, same CLI, same healing loop.

aqueduct run pipeline.yml --set deployment.engine=duckdb

One blueprint format, one CLI, per-engine execution. Each engine declares a capability table, and the compiler checks your blueprint against it. If the target engine can't do something, you get a compile error naming the exact capability instead of a runtime surprise.

Three promises, each enforced by machinery rather than review discipline:

• Every supported entry in the capability matrix is backed by a named test that runs on that engine in CI.
• Nothing that can change your results is ever silently ignored. The engine honors it, warns about it, or refuses to compile.
• Healed blueprints record which engine produced each patch. Deploy a DuckDB-healed blueprint to Spark and the compiler tells you.

The two engines do not promise identical values, and Aqueduct does not pretend they do. SQL dialects genuinely differ; the known differences are cataloged in the Compatibility Matrix. Your Assert rules run on both engines, so the properties you care about are the properties that get checked. That is your portability contract.

Polyglot pipelines

Engines can also mix inside one blueprint. Set engine: on a module and everything downstream inherits it until another module overrides:

modules:
  - id: extract        # heavy join across two warehouses
    type: Channel
    engine: spark
    config: { op: sql, query: "..." }

  - id: aggregate      # small result, no cluster needed
    type: Channel
    engine: duckdb
    config: { op: sql, query: "SELECT region, sum(amount) FROM extract GROUP BY region" }

The compiler partitions the DAG into engine islands and inserts a handoff at each boundary. The handoff materializes data as parquet at a configurable location, visible in observability like any other module, with bytes and duration recorded. Each handoff point is announced at compile time, because the extra I/O is a real cost you should see before the run. If an island fails, a rerun picks up the already-materialized upstream data instead of recomputing it. Two independent flows with different engines run side by side with no handoff at all.

What you get

Core concepts

Full details in the References.

The healing flow

When a pipeline fails, Aqueduct does not throw a stack trace at an LLM and hope. Healing is a staged, auditable pipeline. The model works inside a constrained grammar: it cannot write code, edit files, or run shell commands.

A generated patch clears five gates before it ever touches the Blueprint. Guardrails first: deterministic policy checks on paths, operations, and confidence. Then compile-check: the patched Blueprint must still parse. Then lineage: does the patch break a downstream column consumer. Then sandbox: replay against representative data. Then a plan-regression check. They run in that order and the first failure wins:

✓ guardrails  →  ✓ compile-check  →  ✓ lineage  →  ✓ sandbox  →  ✓ plan-regression  →  patch applied

Every patch clears the pyramid before it touches the Blueprint. aqueduct patch preview --sandbox runs the same pyramid on demand, before you decide to apply.

Approval modes

Who applies a generated patch. Deterministic guardrails (allowed paths, forbidden operations, minimum confidence) bound every patch regardless of mode.

Low-confidence patches and any guardrail violation auto-escalate to human review.

// patches/pending/hello-pipeline-20260611T031412.json (abridged)
{
  "patch_id": "hello-pipeline-20260611T031412",
  "run_id": "9f3c2e1a",
  "category": "config_error",
  "root_cause": "Ingress 'load' reads data/in.csv, but the upstream job renamed the file to data/input.csv.",
  "confidence": 0.92,
  "rationale": "PATH_NOT_FOUND on data/in.csv; a sibling data/input.csv exists with a matching schema, so the path is stale rather than the data missing.",
  "operations": [
    { "op": "set_module_config_key", "module_id": "load", "key": "path", "value": "data/input.csv" }
  ]
}

Why it holds up

• Every change is visible. A patch is a structured diff with a rationale and a confidence score. Low confidence escalates to a human.
• Live data stays safe. The sandbox validates each patch against representative data before any live write.
• Loops are bounded. A multi-axis budget caps wall-clock time, tokens, reprompts, and stuck-signature windows. A rolling rate limit caps heals per hour per blueprint.
• Decisions are auditable. Every LLM turn is recorded with the gate that rejected it, a stable error signature, and the prompt version. One run id joins every iteration of a heal.
• Efficient. Healing stops on the first successful patch. Structured error extraction replaces multi-kilobyte traces with a short root-cause block. Cheap lineage and sandbox checks reject bad patches in seconds, before any full-pipeline replay.
• Reach beyond the blueprint, with the same discipline. Healing extends past pipeline definitions into engine and session config (allowlisted keys only) and dependency declarations, each behind its own gate. Config patches are validated against a per-engine allowlist and replayed in the sandbox. Dependency fixes are declared and validated, then delivered for redeploy rather than live-installed. Data mutation stays off by default: the agent never touches your data without an explicit opt-in and per-action human approval.

For the stage-by-stage detail, see the Blueprint & Engine Spec.

Architecture

Where Aqueduct sits in a data platform: the control plane between your scheduler and the engine.

Inside the box, Aqueduct is a single CLI that runs on the driver, with no servers and no daemons. Logic flows through four immutable layers:

• Parser validates YAML into an immutable AST.
• Compiler resolves context, expands Arcades and macros, extracts column lineage, gates the blueprint against the target engine's capabilities, and assembles a fully-resolved Manifest.
• Executor runs the Manifest on the target engine. Engines register through an entry-point protocol; Spark code is isolated under executor/spark/, DuckDB under executor/duckdb_/. The core never imports an engine by name.
• Surveyor records runs, failures, and lineage to pluggable stores and triggers the Agent on failure.

Observability dashboard

aqueduct dashboard launches a local, read-only Streamlit viewer over the same observability store the engine writes to. It runs on demand, like the Spark UI. It is not a production server, and no pipeline requires it. One place for fleet health, per-run module metrics, column lineage, the self-heal patch stream with before/after diffs, performance trends, and data-quality signals across every blueprint. Backend-agnostic (DuckDB or Postgres). Every view re-reads with short-lived connections, so it can't block a running pipeline's writer.

pip install "aqueduct-core[dashboard]"
aqueduct dashboard            # opens http://localhost:8501

Getting started

Installation

pip install aqueduct-core              # DuckDB engine included, no JVM needed
pip install "aqueduct-core[spark]"     # adds Apache Spark + Delta Lake

Requirements: Python 3.11+. Java 17 for the spark extra only (JAVA_HOME must point to it). Every release is CI-tested against three pinned combos: LTS (Python 3.11 · Spark 4.1), Latest (Python 3.13 · Spark 4.1), Legacy (Python 3.12 · Spark 3.5). Live results in the Compatibility Matrix.

Compose extras as needed, for example pip install "aqueduct-core[spark,airflow,aws]":

A first blueprint

aqueduct: "1.0"
id: hello.pipeline
name: Hello Pipeline

macros:
  active: "status = 'active' AND deleted_at IS NULL"

modules:
  - id: load
    type: Ingress
    label: Load orders
    config: { format: csv, path: "data/in.csv", options: { header: true } }

  - id: clean
    type: Channel
    label: Filter active
    config:
      op: sql
      query: "SELECT order_id, amount FROM load WHERE {{ macros.active }}"

  - id: save
    type: Egress
    label: Write parquet
    config: { format: parquet, path: "data/out/", mode: overwrite }

edges:
  - { from: load,  to: clean }
  - { from: clean, to: save }

agent:
  approval: human

Run it on either engine:

aqueduct run blueprints/hello.yml                                  # engine from aqueduct.yml
aqueduct run blueprints/hello.yml --set deployment.engine=duckdb   # no cluster, no JVM

Engine-wide defaults live in a separate aqueduct.yml (target engine, LLM provider, store backends, danger settings). Inline module tests live in *.aqtest.yml. Repeatable healing benchmarks live in *.aqscenario.yml. The Gallery has runnable examples of each.

Five commands to know

1. aqueduct doctor blueprints/hello.yml is the preflight check. It validates YAML, resolves paths, verifies LLM reachability, and opens stores.
2. aqueduct run blueprints/hello.yml executes the pipeline. On failure, the agent generates a patch under patches/pending/.
3. aqueduct patch apply patches/pending/<id>.json --blueprint blueprints/hello.yml reviews and accepts a staged patch, moving it to patches/applied/.
4. aqueduct test blueprints/hello.aqtest.yml runs Channel / Junction / Funnel modules against inline data, without touching Ingress, Egress, or any external I/O.
5. aqueduct benchmark gallery/aqscenarios/ --model claude-sonnet-4-6 --model qwen2.5-coder:7b compares LLM models against simulated failures. No engine required.

Full reference in CLI Reference.

How it compares

Aqueduct sits where a transformation engine and an autonomous repair loop meet. It is not a scheduler (pair it with Airflow via the airflow extra) and it is not a warehouse SQL tool.

References

• Blueprint & Engine Spec: module types, configs, architecture, type system, healing loop
• SKILL.md: distilled Blueprint-authoring guide for LLMs (grammar, patterns, provider base_urls)
• CLI Reference: all commands and flags
• Spark Engine Guide: warnings, performance, tuning
• Observability Guide: schemas + diagnostic query cookbook
• Production Guide: cluster deployment, security, Delta operations
• Compatibility Matrix: supported Python × Spark versions, per-engine capability tables
• Extending Aqueduct: how to add an execution engine (ExecutorProtocol, capability declarations, entry points)
• Roadmap: deferred features and future plans
• Gallery: real working examples

Contributing

Contributions are welcome! See CONTRIBUTING.md.

Aqueduct is Apache 2.0 licensed: free and open source, with no telemetry and no lock-in.