Motherduck Build Data Pipeline

Build a Data Pipeline with MotherDuck

Use this skill when the user needs an ingestion-to-serving workflow, not just a single load step.

This is a use-case skill. It orchestrates motherduck-connect, motherduck-load-data, motherduck-model-data, motherduck-query, motherduck-share-data, and motherduck-ducklake.

Start Here: Is a MotherDuck Server Active?

Always determine this first.

• If a remote MotherDuck MCP server or local MotherDuck server is active, use it.
• If the user already knows the destination database, confirm it before designing stages.
• Explore the live environment:
• current databases and schemas
• raw, staging, and analytics boundaries if they already exist
• source tables, target tables, and table grain
• key columns, date fields, and join keys

Use that discovery to decide whether the pipeline is:

• landing into an empty workspace
• extending an existing warehouse layout
• publishing into an existing analytics model

If no server is active, ask for source shape and target shape before drafting the pipeline.

Use This Skill When

• The user needs ingestion plus transformation plus serving output.
• The work spans raw landing, curation, and publication.
• The user needs a stage-by-stage pipeline pattern rather than one command.
• The problem is bigger than a single import step or one ad hoc transformation.

Pipeline Defaults

• batch over streaming
• raw landing before curation
• explicit raw -> staging -> analytics boundaries
• bulk ingest paths over row-by-row writes
• idempotent stage rebuilds or append contracts before scheduled automation
• verify the MotherDuck-supported DuckDB client version before recommending upstream-only write, checkpoint, or lakehouse features
• native MotherDuck storage unless DuckLake is explicitly required

Workflow

1. Confirm whether live MotherDuck discovery is available. 2. Inspect the current workspace and target data model. 3. Define raw, staging, and analytics boundaries. 4. Ingest raw data. 5. Deduplicate, type, and promote into staging. 6. Materialize analytics-ready outputs. 7. Validate counts, freshness, uniqueness, and business metrics before publishing downstream assets.

When this skill produces a native DuckDB (md:) connection, watermark it with custom_user_agent=agent-skills/2.3.0(harness-<harness>;llm-<llm>). If metadata is missing, fall back to harness-unknown and llm-unknown.

Output

The output of this skill should be:

• the stage layout
• the ingestion method
• the transformation sequence
• the serving tables or views
• the validation checks

If the caller explicitly asks for structured JSON, return raw JSON only with no Markdown fences or prose before/after it. This is mainly for automated tests, regression checks, or downstream tooling that needs a stable machine-readable shape. Normal human-facing use of the skill can stay in prose unless JSON is explicitly requested.

Use this exact top-level shape when JSON is requested:

{
  "summary": {},
  "assumptions": [],
  "implementation_plan": [],
  "validation_plan": [],
  "risks": []
}

References

• references/dlt-dbt-motherduck-project/ -- fully runnable MotherDuck reference project using dlt, dbt-duckdb, and validation queries
• references/PIPELINE_IMPLEMENTATION_GUIDE.md -- preserved detailed pipeline guidance that used to live in this skill
• ../motherduck-load-data/references/INGESTION_PATTERNS.md -- lower-level ingestion patterns

Runnable Artifact

• artifacts/pipeline_stage_example.py -- MotherDuck-backed Python example that stages a Parquet extract, lands it into raw, deduplicates it, and publishes analytics output across raw/staging/analytics databases
• artifacts/pipeline_stage_example.ts -- TypeScript companion artifact with the same stage layout and output contract
• references/dlt-dbt-motherduck-project/ -- end-to-end MotherDuck example that bootstraps the target database, lands raw data with dlt, builds staging and analytics models with dbt, and validates the final mart

Run it with:

uv run --with duckdb python skills/motherduck-build-data-pipeline/artifacts/pipeline_stage_example.py

Run the same stage pattern against temporary MotherDuck databases:

MOTHERDUCK_ARTIFACT_USE_MOTHERDUCK=1 \
uv run --with duckdb python skills/motherduck-build-data-pipeline/artifacts/pipeline_stage_example.py

Validate the TypeScript companion artifact:

uv run scripts/test_typescript_artifacts.py

For the full MotherDuck project:

cd skills/motherduck-build-data-pipeline/references/dlt-dbt-motherduck-project
export MOTHERDUCK_TOKEN=...
export MOTHERDUCK_PIPELINE_DB=md_skills_pipeline_demo
uv sync --python 3.12
uv run python pipeline/run_all.py
uv run python pipeline/cleanup.py

Verified Notes

• Bootstrap the target MotherDuck database before running dlt. The motherduck destination does not create the database for you.
• Keep this stack on Python 3.11 or 3.12 for now. The tested dbt-duckdb path here was not reliable on Python 3.14.
• If you want exact schema names like raw, staging, and analytics in dbt, override generate_schema_name.
• When a long-lived Python process loads data and a separate dbt subprocess builds models, run post-build validation in a fresh process or refresh database state before reading new relations.

Related Skills

• motherduck-connect -- choose the right connection path
• motherduck-load-data -- ingestion mechanics
• motherduck-model-data -- shape the analytics layer
• motherduck-query -- write transformations and validations
• motherduck-share-data -- publish curated outputs
• motherduck-ducklake -- only when open-table-format storage is a real requirement

dlt + dbt + MotherDuck Reference Project

This is a minimal end-to-end pipeline reference for the motherduck-build-data-pipeline skill.

It captures the pipeline shape that this repo repeatedly verified against real MotherDuck runs:

• dlt for the raw loading step
• dbt-duckdb for staging and analytics modeling
• Python validation and cleanup around the workflow

The example is deliberately small and fully runnable:

• dlt lands raw JSONL data in MotherDuck
• dbt models staging and analytics relations in the same MotherDuck database
• Python validation checks the final outputs

Why One Database

The main skill recommends separate lifecycle stages. For this reference project, the simplest runnable shape is one MotherDuck database with explicit schemas:

• raw
• staging
• analytics

That keeps the dbt project small and avoids extra attach configuration. When the pipeline grows and stage boundaries matter operationally, split the stages into separate MotherDuck databases and use dbt attach.

Verified Constraints

These are based on a real local run against MotherDuck:

• Bootstrap the MotherDuck database before dlt runs. The motherduck destination does not create the target database for you.
• Use Python 3.11 or 3.12 for this stack. dbt-duckdb did not run correctly on Python 3.14 in this environment.
• Keep dbt concurrency at threads: 1 for a small MotherDuck project like this.
• Override generate_schema_name so dbt uses exact schema names instead of main_<schema>.
• Run post-build validation in a fresh process. A long-lived local DuckDB process may not immediately see schemas written by a separate dbt subprocess.

Files

• pipeline/bootstrap.py: creates the MotherDuck database and schemas
• pipeline/load_raw.py: loads raw data into MotherDuck with dlt
• pipeline/run_all.py: runs bootstrap, load, dbt build, and validation
• pipeline/validate.py: asserts row counts and final mart output
• dbt_project.yml, profiles.yml, models/, macros/: dbt project
• data/*.jsonl: tiny input dataset

Run It

Set credentials:

export MOTHERDUCK_TOKEN=...
export MOTHERDUCK_PIPELINE_DB=md_skills_pipeline_demo

Install dependencies with a supported Python:

uv sync --python 3.12

Run the whole pipeline:

uv run python pipeline/run_all.py

Drop the temporary MotherDuck database when you are done:

uv run python pipeline/cleanup.py

Run the steps individually if you want to inspect them:

uv run python pipeline/bootstrap.py
uv run python pipeline/load_raw.py
DBT_PROFILES_DIR=. uv run dbt build
uv run python pipeline/validate.py

Expected Output

After a successful run, the final mart contains three rows:

The staging model also proves two common pipeline patterns:

• deduplicate by latest updated_at
• filter analytics output to PAID orders only

Real MotherDuck Test Posture

This project is intended to run against a real MotherDuck database, not just local DuckDB.

• use a temporary MOTHERDUCK_PIPELINE_DB for validation runs
• bootstrap the database first
• run validation after dbt build
• drop the temporary database with pipeline/cleanup.py after the run

<!-- Preserved detailed implementation guidance moved from SKILL.md so the main skill can stay concise. -->

Build a Data Pipeline with MotherDuck

Use this skill when designing an end-to-end workflow that moves data from raw sources through transformation stages into analytics-ready output. This is a use-case skill -- it ties together lower-level skills into a complete pipeline.

Contents

• Source Of Truth
• Language Focus: TypeScript/Javascript and Python
• TypeScript/Javascript Orchestration Starter
• Prerequisites
• Runnable Reference Project
• Verified Delivery Defaults
• Validation Signals
• Pipeline Architecture
• Step 1: Design the Target Schema
• Step 2: Ingest Raw Data into Raw
• Step 3: Promote Into Staging and Write Transformation Queries
• Step 4: Materialize Analytics Tables
• Step 5: Validate Data Quality
• Step 6: Serve Results
• Incremental Load Patterns
• Complete Pipeline Example
• Scheduling Considerations
• Key Rules
• Common Mistakes
• Related Skills

Source Of Truth

• Prefer current MotherDuck loading, connection, tagging, and storage docs first.
• If the MotherDuck MCP ask_docs_question feature is available, use it before falling back to public docs.
• Keep the pipeline guidance aligned with the documented posture:
• batch over streaming
• raw landing before curation
• Parquet and bulk paths over row-by-row inserts
• native MotherDuck storage first unless DuckLake is explicitly required

Language Focus: TypeScript/Javascript and Python

• Prefer Python as the default language for pipeline implementation:
• ingestion jobs
• transformation runners
• notebook validation
• orchestration glue
• Prefer TypeScript/Javascript when the pipeline connects directly to:
• backend services
• event ingestion APIs
• product-side control planes
• If the user asks for implementation code, bias toward Python unless their existing stack is clearly Node.js.

TypeScript/Javascript Orchestration Starter

For Node.js pipelines, prefer the native DuckDB path when you need any of these:

• local-file ingestion
• extension-backed reads
• hybrid local and remote execution
• tighter control over DuckDB behavior

Use the PG endpoint only when the pipeline already lives in a PostgreSQL-driver environment and the work is limited to server-side SQL against MotherDuck-managed data or remote object reads.

Native DuckDB path for Node.js:

import { DuckDBInstance } from "@duckdb/node-api";
import { readFile } from "node:fs/promises";

const instance = await DuckDBInstance.create(
  "md:?custom_user_agent=agent-skills/2.3.0(harness-<harness>;llm-<llm>)"
);
const conn = await instance.connect();
for (const file of ["01_ingest.sql", "02_transform.sql", "03_publish.sql"]) {
  await conn.run(await readFile(`sql/pipeline/${file}`, "utf8"));
}
conn.close();

PG endpoint path for existing PostgreSQL-driver stacks:

import pg from "pg";
import { readFile } from "node:fs/promises";

const client = new pg.Client({
  host: "pg.us-east-1-aws.motherduck.com",
  port: 5432,
  database: "staging",
  user: "postgres",
  password: process.env.MOTHERDUCK_TOKEN,
  ssl: { rejectUnauthorized: true },
});

await client.connect();
for (const file of ["01_ingest.sql", "02_transform.sql", "03_publish.sql"]) {
  await client.query(await readFile(`sql/pipeline/${file}`, "utf8"));
}
await client.end();

Do not use the PG endpoint for local-file COPY, extension installation, or other client-only DuckDB behaviors.

Prerequisites

• MotherDuck connection established (see motherduck-connect skill)
• Familiarity with data ingestion patterns (see motherduck-load-data skill)
• Understanding of schema design (see motherduck-model-data skill)
• Ability to write transformation queries (see motherduck-query skill)

Runnable Reference Project

For a fully runnable example in this repo, start with:

• references/dlt-dbt-motherduck-project/

That reference project is intentionally small and verified against a real MotherDuck run. It combines:

• dlt for raw loading
• dbt-duckdb for staging and analytics models
• Python validation for output checks

Operational notes from that verified example:

• bootstrap the target MotherDuck database before running dlt; the motherduck destination does not create the database for you
• keep this stack on Python 3.11 or 3.12 for now; the tested dbt-duckdb path here was not reliable on Python 3.14
• if you want exact schema names like raw, staging, and analytics in dbt, override generate_schema_name; otherwise dbt defaults may append the target schema name

Verified Delivery Defaults

The repeated repo runs point to a stable pipeline posture:

• prefer Parquet or other bulk landing paths over row inserts
• keep explicit raw, staging, and analytics boundaries even in small examples
• ship one small MotherDuck-backed artifact plus one deeper runnable reference project
• measure and validate the pipeline with real MotherDuck runs rather than relying on local-only examples
• bootstrap the target MotherDuck database before loaders that assume it already exists

Validation Signals

Use these signals for testing, review, and regression checks. They are not an instruction to include a separate "Validation Signals" section in normal user-facing replies.

• run artifacts/pipeline_stage_example.py against temporary MotherDuck databases
• verify the output reports ingestion_mode as bulk_parquet_stage
• verify the stage counts show raw > staging only when deduplication is expected
• run references/dlt-dbt-motherduck-project/ end to end when the change affects the reference pipeline shape

---

Pipeline Architecture

Every pipeline follows four stages. Do not skip stages.

For code examples and execution:

• default to Python for the pipeline runner
• show TypeScript/Javascript only when the pipeline is embedded in an existing Node.js service or control plane

Source --> Raw --> Staging --> Analytics/Serve

• Source: External data -- files, cloud storage (S3, GCS, Azure), APIs, databases.
• Raw: Append-only landing data preserved as close to the source as practical.
• Staging: Cleaned, typed, deduplicated intermediate tables.
• Analytics/Serve: Analytics-ready tables, views, Dives, or shares for downstream consumption.

Separating stages ensures you never lose raw data, can debug transformations independently, and can rebuild downstream assets from raw or staging at any time.

When stages live in separate databases, MotherDuck supports cross-database queries seamlessly. Reference tables in other databases with fully qualified names:

-- Query staging data from the analytics database context
SELECT * FROM "raw"."main"."orders_landing" WHERE order_date >= '2024-01-01';
-- Join across databases
SELECT s.*, c.customer_name
FROM "staging"."main"."orders_clean" s
LEFT JOIN "raw"."main"."customers_landing" c ON s.customer_id = c.customer_id;

This means pipeline SQL does not need to switch database context between stages -- every query can reference any stage by name.

---

Step 1: Design the Target Schema

Start from the end -- what does the analytics team need? Design output tables first, then work backward.

For production pipelines, prefer a multi-database structure to enforce stage separation:

CREATE DATABASE IF NOT EXISTS raw;       -- Append-only ingested data
CREATE DATABASE IF NOT EXISTS staging;   -- Cleaned and deduplicated data
CREATE DATABASE IF NOT EXISTS analytics; -- Denormalized, business-ready tables

Design wide, denormalized analytics tables (see motherduck-model-data skill). Pre-join dimensions so analysts do not need to write joins.

For a minimal dbt project, one MotherDuck database with explicit raw, staging, and analytics schemas is also acceptable. That keeps the project small while preserving stage boundaries in the relation names.

---

Step 2: Ingest Raw Data into Raw

Use motherduck-load-data skill patterns. Land data in raw as-is -- no transformations at this stage.

CREATE OR REPLACE TABLE "raw"."main"."orders_landing" AS
SELECT * FROM read_parquet('s3://bucket/orders/*.parquet');

CREATE OR REPLACE TABLE "raw"."main"."customers_landing" AS
SELECT * FROM read_csv('s3://bucket/customers/customers.csv');

Use CREATE OR REPLACE TABLE for idempotent full refreshes. Validate after loading:

SELECT 'orders_landing' AS table_name, count(*) AS row_count FROM "raw"."main"."orders_landing"
UNION ALL
SELECT 'customers_landing', count(*) FROM "raw"."main"."customers_landing";

Operational defaults:

• buffer API or event traffic before writing analytical tables
• prefer staged Parquet, Arrow/dataframes, or COPY
• tag long-lived workloads with custom_user_agent; for repo use-case builds, use agent-skills/2.3.0(harness-<harness>;llm-<llm>)
• keep write transactions comfortably bounded instead of unbounded monoliths

---

Step 3: Promote Into Staging and Write Transformation Queries

Apply transformations in order: deduplicate, cast types, join, aggregate. Use CTEs for readability.

Deduplication

CREATE OR REPLACE TABLE "staging"."main"."orders_deduped" AS
SELECT * FROM "raw"."main"."orders_landing"
QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) = 1;

For composite keys, partition by the full key:

CREATE OR REPLACE TABLE "staging"."main"."order_lines_deduped" AS
SELECT * FROM "raw"."main"."order_lines_landing"
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY order_id, line_item_id
    ORDER BY updated_at DESC
) = 1;

Type Casting and Normalization

CREATE OR REPLACE TABLE "staging"."main"."orders_clean" AS
SELECT
    order_id,
    CAST(order_date AS DATE) AS order_date,
    customer_id,
    CAST(quantity AS INTEGER) AS quantity,
    CAST(unit_price AS DECIMAL(18,2)) AS unit_price,
    CAST(quantity * unit_price AS DECIMAL(18,2)) AS total_amount,
    UPPER(TRIM(status)) AS status
FROM "staging"."main"."orders_deduped"
WHERE order_id IS NOT NULL;

Joining Across Sources

CREATE OR REPLACE TABLE "analytics"."main"."orders" AS
SELECT
    o.order_id, o.order_date, o.customer_id,
    c.customer_name, c.segment AS customer_segment,
    p.product_name, p.category AS product_category,
    o.quantity, o.unit_price, o.total_amount, c.region
FROM "staging"."main"."orders_clean" o
LEFT JOIN "raw"."main"."customers_landing" c ON o.customer_id = c.customer_id
LEFT JOIN "raw"."main"."products_landing" p ON o.product_id = p.product_id;
COMMENT ON TABLE "analytics"."main"."orders" IS 'Denormalized order data with customer and product attributes';

Aggregation

CREATE OR REPLACE TABLE "analytics"."main"."daily_revenue" AS
SELECT
    order_date, region, product_category,
    COUNT(*) AS order_count, SUM(total_amount) AS total_revenue,
    AVG(total_amount) AS avg_order_value, COUNT(DISTINCT customer_id) AS unique_customers
FROM "analytics"."main"."orders"
GROUP BY ALL;
COMMENT ON TABLE "analytics"."main"."daily_revenue" IS 'Daily revenue by region and product category from orders';
COMMENT ON COLUMN "analytics"."main"."daily_revenue"."total_revenue" IS 'SUM(total_amount) from analytics.main.orders';
COMMENT ON COLUMN "analytics"."main"."daily_revenue"."avg_order_value" IS 'AVG(total_amount) from analytics.main.orders';
COMMENT ON COLUMN "analytics"."main"."daily_revenue"."unique_customers" IS 'COUNT(DISTINCT customer_id) from analytics.main.orders';

---

Step 4: Materialize Analytics Tables

Use CTAS for expensive aggregations queried repeatedly. Use views for lightweight, always-current logic.

-- Materialized: expensive computation
CREATE OR REPLACE TABLE "analytics"."main"."customer_lifetime_value" AS
SELECT
    customer_id, customer_name, customer_segment,
    COUNT(DISTINCT order_id) AS total_orders,
    SUM(total_amount) AS lifetime_revenue,
    MIN(order_date) AS first_order_date,
    MAX(order_date) AS last_order_date
FROM "analytics"."main"."orders"
GROUP BY ALL;
COMMENT ON TABLE "analytics"."main"."customer_lifetime_value" IS 'Customer lifetime value metrics from orders';
COMMENT ON COLUMN "analytics"."main"."customer_lifetime_value"."total_orders" IS 'COUNT(DISTINCT order_id) from analytics.main.orders';
COMMENT ON COLUMN "analytics"."main"."customer_lifetime_value"."lifetime_revenue" IS 'SUM(total_amount) from analytics.main.orders';
COMMENT ON COLUMN "analytics"."main"."customer_lifetime_value"."first_order_date" IS 'MIN(order_date) from analytics.main.orders';
COMMENT ON COLUMN "analytics"."main"."customer_lifetime_value"."last_order_date" IS 'MAX(order_date) from analytics.main.orders';

-- View: always-current, lightweight
CREATE OR REPLACE VIEW "analytics"."main"."recent_orders" AS
SELECT * FROM "analytics"."main"."orders"
WHERE order_date >= current_date - INTERVAL 30 DAY;
COMMENT ON VIEW "analytics"."main"."recent_orders" IS 'Orders from the last 30 days from analytics.main.orders';

---

Step 5: Validate Data Quality

Run validation checks between every pipeline stage. Never skip validation.

-- Row count sanity check across stages
SELECT 'raw.orders_landing' AS table_name, count(*) AS row_count
    FROM "raw"."main"."orders_landing"
UNION ALL
SELECT 'staging.orders_deduped', count(*) FROM "staging"."main"."orders_deduped"
UNION ALL
SELECT 'analytics.orders', count(*) FROM "analytics"."main"."orders";

-- NULL check on required columns
SELECT
    count(*) FILTER (WHERE order_id IS NULL) AS null_order_ids,
    count(*) FILTER (WHERE customer_id IS NULL) AS null_customer_ids,
    count(*) FILTER (WHERE total_amount IS NULL) AS null_amounts
FROM "analytics"."main"."orders";

-- Uniqueness check
SELECT order_id, count(*) AS cnt FROM "analytics"."main"."orders"
GROUP BY order_id HAVING cnt > 1;

-- Range validation
SELECT MIN(order_date) AS earliest, MAX(order_date) AS latest,
    count(*) FILTER (WHERE total_amount < 0) AS negative_amounts
FROM "analytics"."main"."orders";

---

Step 6: Serve Results

-- Views for common query patterns
CREATE OR REPLACE VIEW "analytics"."main"."top_customers" AS
SELECT customer_id, customer_name, lifetime_revenue
FROM "analytics"."main"."customer_lifetime_value"
ORDER BY lifetime_revenue DESC LIMIT 100;
COMMENT ON VIEW "analytics"."main"."top_customers" IS 'Top 100 customers by lifetime revenue from customer_lifetime_value';

• Use the motherduck-create-dive skill for interactive visualizations powered by analytics tables.
• Use the motherduck-share-data skill to distribute databases to teams or partners:

CREATE SHARE IF NOT EXISTS analytics_share FROM analytics (
    ACCESS ORGANIZATION, VISIBILITY DISCOVERABLE, UPDATE AUTOMATIC
);

Before sharing, make sure the serving tables are curated and documented. Shares are zero-copy and easy to distribute, so be deliberate about what database boundary you are publishing.

---

Incremental Load Patterns

Full refreshes work for small-to-medium datasets. For large or frequently updated datasets, use incremental patterns.

-- Append new data only
INSERT INTO "raw"."main"."orders_landing"
SELECT * FROM read_parquet('s3://bucket/orders/date=2024-03-24/*.parquet')
WHERE order_date > (SELECT MAX(order_date) FROM "raw"."main"."orders_landing");

-- Upsert: load into temp table, delete old rows, insert new
CREATE OR REPLACE TEMP TABLE new_orders AS
SELECT * FROM read_parquet('s3://bucket/orders/latest/*.parquet');

DELETE FROM "raw"."main"."orders_landing"
WHERE order_id IN (SELECT order_id FROM new_orders);

INSERT INTO "raw"."main"."orders_landing"
SELECT * FROM new_orders;

-- Incremental aggregation: rebuild only affected date range
DELETE FROM "analytics"."main"."daily_revenue"
WHERE order_date >= (SELECT MAX(order_date) - INTERVAL 3 DAY FROM "raw"."main"."orders_landing");

INSERT INTO "analytics"."main"."daily_revenue"
SELECT order_date, region, product_category,
    COUNT(*) AS order_count, SUM(total_amount) AS total_revenue,
    AVG(total_amount) AS avg_order_value, COUNT(DISTINCT customer_id) AS unique_customers
FROM "analytics"."main"."orders"
WHERE order_date >= (SELECT MAX(order_date) - INTERVAL 3 DAY FROM "raw"."main"."orders_landing")
GROUP BY ALL;

---

Complete Pipeline Example

End-to-end: CSV ingest, deduplicate, join, aggregate, validate, create views, share.

-- 1. Create databases
CREATE DATABASE IF NOT EXISTS raw;
CREATE DATABASE IF NOT EXISTS staging;
CREATE DATABASE IF NOT EXISTS analytics;

-- 2. Ingest raw data
CREATE OR REPLACE TABLE "raw"."main"."sales_landing" AS
SELECT * FROM read_csv('s3://acme-data/sales/sales_2024.csv');

CREATE OR REPLACE TABLE "raw"."main"."customers_landing" AS
SELECT * FROM read_csv('s3://acme-data/customers/customers.csv');

-- 3. Deduplicate
CREATE OR REPLACE TABLE "staging"."main"."sales_deduped" AS
SELECT * FROM "raw"."main"."sales_landing"
QUALIFY ROW_NUMBER() OVER (PARTITION BY sale_id ORDER BY updated_at DESC) = 1;

-- 4. Transform and join
CREATE OR REPLACE TABLE "analytics"."main"."sales" AS
SELECT s.sale_id, s.sale_date, s.product_name, s.quantity, s.unit_price,
    CAST(s.quantity * s.unit_price AS DECIMAL(18,2)) AS total_amount,
    c.customer_name, c.segment, c.region
FROM "staging"."main"."sales_deduped" s
LEFT JOIN "raw"."main"."customers_landing" c ON s.customer_id = c.customer_id;
COMMENT ON TABLE "analytics"."main"."sales" IS 'Denormalized sales with customer attributes';

-- 5. Aggregate
CREATE OR REPLACE TABLE "analytics"."main"."revenue_summary" AS
SELECT date_trunc('month', sale_date) AS month, region, segment,
    COUNT(*) AS sale_count, SUM(total_amount) AS total_revenue,
    COUNT(DISTINCT customer_name) AS unique_customers
FROM "analytics"."main"."sales" GROUP BY ALL;
COMMENT ON TABLE "analytics"."main"."revenue_summary" IS 'Monthly revenue summary by region and segment from sales';
COMMENT ON COLUMN "analytics"."main"."revenue_summary"."total_revenue" IS 'SUM(total_amount) from analytics.main.sales';
COMMENT ON COLUMN "analytics"."main"."revenue_summary"."unique_customers" IS 'COUNT(DISTINCT customer_name) from analytics.main.sales';

-- 6. Validate
SELECT count(*) FILTER (WHERE sale_id IS NULL) AS null_ids,
    count(*) FILTER (WHERE total_amount < 0) AS negative_amounts
FROM "analytics"."main"."sales";

SELECT sale_id, count(*) AS cnt FROM "analytics"."main"."sales"
GROUP BY sale_id HAVING cnt > 1;

-- 7. Serve
CREATE OR REPLACE VIEW "analytics"."main"."monthly_revenue" AS
SELECT month, SUM(total_revenue) AS revenue, SUM(unique_customers) AS customers
FROM "analytics"."main"."revenue_summary" GROUP BY month ORDER BY month;
COMMENT ON VIEW "analytics"."main"."monthly_revenue" IS 'Monthly total revenue and customer counts rolled up from revenue_summary';
COMMENT ON COLUMN "analytics"."main"."monthly_revenue"."revenue" IS 'SUM(total_revenue) from analytics.main.revenue_summary';
COMMENT ON COLUMN "analytics"."main"."monthly_revenue"."customers" IS 'SUM(unique_customers) from analytics.main.revenue_summary';

-- 8. Share
CREATE SHARE IF NOT EXISTS analytics_share FROM analytics (
    ACCESS ORGANIZATION, VISIBILITY DISCOVERABLE, UPDATE AUTOMATIC
);

---

Scheduling Considerations

MotherDuck does not have built-in scheduling. Use external schedulers: cron, GitHub Actions, Dagster, Airflow, or Prefect.

Store SQL transformations in version-controlled .sql files. Execute them from a scheduled script.

Native DuckDB (recommended)

Use native duckdb.connect("md:") for pipeline runners. This gives you full DuckDB SQL support, cross-database queries, and no driver translation layer.

# pipeline.py -- run via cron, Airflow, or GitHub Actions
import duckdb
import os
from pathlib import Path

PIPELINE_USER_AGENT = "agent-skills/2.3.0(harness-<harness>;llm-<llm>)"

def run_pipeline():
    conn = duckdb.connect(f"md:?custom_user_agent={PIPELINE_USER_AGENT}")
    for step in sorted(Path("sql/pipeline").glob("*.sql")):
        print(f"Running {step.name}...")
        conn.execute(step.read_text())
    conn.close()

if __name__ == "__main__":
    run_pipeline()

PG endpoint alternative

Use the PG endpoint when the pipeline runs in an environment that already has PostgreSQL drivers and you want to avoid installing duckdb. This is common in serverless runtimes, container images with existing psycopg2, or TypeScript backends.

# pipeline_pg.py -- PG endpoint alternative
import psycopg2, certifi, os
from pathlib import Path

def run_pipeline():
    conn = psycopg2.connect(
        host="pg.us-east-1-aws.motherduck.com", port=5432,
        dbname="staging", user="postgres",
        password=os.environ["MOTHERDUCK_TOKEN"],
        sslmode="verify-full", sslrootcert=certifi.where(),
    )
    conn.autocommit = True
    for step in sorted(Path("sql/pipeline").glob("*.sql")):
        print(f"Running {step.name}...")
        conn.cursor().execute(step.read_text())
    conn.close()

if __name__ == "__main__":
    run_pipeline()

Number files to enforce execution order (01_ingest.sql, 02_dedupe.sql, etc.). Each file should be idempotent -- use CREATE OR REPLACE so re-running is safe.

---

Key Rules

• Separate lifecycle stages explicitly. Production default: raw, staging, and analytics as separate databases. Minimal dbt projects may use one database with raw, staging, and analytics schemas.
• Land data in `raw` before curation. Preserve source-like tables so downstream rebuilds stay simple.
• Validate data between every pipeline stage. Row counts, NULL checks, uniqueness, range validation.
• Preserve raw data. Never transform during ingestion. Rebuild downstream tables from staging.
• Materialize only what needs fast repeated access. Use views for lightweight, always-current logic.
• Use `CREATE OR REPLACE` for idempotent rebuilds. Every pipeline step should be safe to re-run.
• Version control all SQL transformations. Store .sql files in git, not in ad-hoc query editors.
• Deduplicate before building analytics tables. Raw sources often contain duplicates.
• Use fully qualified table names in every statement: "database"."schema"."table".
• Tag long-lived pipeline runners with `custom_user_agent`. This makes workload attribution and cost analysis possible later. For repo use-case builds, use agent-skills/2.3.0(harness-<harness>;llm-<llm>).

---

Common Mistakes

Loading and transforming in a single step

Combining ingestion with transformation loses the raw data. If a bug is discovered later, you must re-ingest from the external source.

-- Wrong: raw data is lost
CREATE TABLE "analytics"."main"."orders" AS
SELECT order_id, UPPER(status) AS status FROM read_parquet('s3://bucket/orders.parquet');

-- Right: ingest raw first, then transform
CREATE TABLE "raw"."main"."orders_landing" AS
SELECT * FROM read_parquet('s3://bucket/orders.parquet');
CREATE TABLE "analytics"."main"."orders" AS
SELECT order_id, UPPER(status) AS status FROM "raw"."main"."orders_landing";

Not deduplicating before analytics tables

Raw sources frequently contain duplicates from retries, overlapping file loads, or CDC replication. Without deduplication, aggregations produce inflated numbers.

Forgetting data validation between stages

Skipping validation means bad data propagates silently. A NULL customer ID in staging becomes an orphaned order in analytics and an incorrect revenue number in a dashboard.

Using DROP TABLE then CREATE TABLE instead of CREATE OR REPLACE

DROP then CREATE is non-atomic -- queries fail during the gap. Use CREATE OR REPLACE TABLE for atomic replacement.

Over-aggregating and losing detail

Pre-aggregating to monthly granularity when analysts later need hourly or daily breakdowns forces a pipeline rebuild. Ask what the finest useful grain is before choosing -- for some use cases that is hourly, for others event-level. Keep that grain as the base table and build coarser rollups (daily, weekly, monthly) on top. When in doubt, preserve more detail -- it is easy to aggregate up but impossible to disaggregate down.

---

Related Skills

• motherduck-connect -- Establish a MotherDuck connection
• motherduck-load-data -- Ingest data from files, cloud storage, and external sources
• motherduck-model-data -- Design database schemas and data models
• motherduck-query -- Execute DuckDB SQL queries and transformations
• motherduck-explore -- Discover databases, tables, columns, and shares
• motherduck-create-dive -- Build interactive visualizations from analytics tables
• motherduck-share-data -- Distribute analytics databases to teams and partners