DuckDB pipelines: COPY and partitioned Parquet

For three monthly exports, a DuckDB pipeline is a short script that turns raw files into data other people can use. It reads the exports with stated types, cleans them into tables, keeps those tables in a .duckdb database file, and publishes the result with COPY … TO as Parquet, often split into one folder per month or per region so that readers open only the part they need. Every step is one SQL statement, so the whole pipeline is easy to read and to re-run. What makes it trustworthy is not the statements but the checks between them: the row counts of every stage must add up, and a second run must not change the answer. Use a pipeline like this for analyst-scale data, millions of rows on one machine, where a warehouse would be more system than the job needs. This lesson builds one from three monthly CSV exports, counts every stage, and names the statements and options a pipeline uses.

Stage counts explain where all sixteen lines went

Three monthly exports of invoice lines arrive in raw/. July’s has one line exported twice, August’s has a date typed as 2026-08-3O with a letter O, and September’s includes a refund:

invoice,line,sold_on,city,sku,qty,amount
INV-701,1,2026-07-03,Pune,BK-101,2,900.00
INV-701,2,2026-07-03,Pune,BK-102,1,299.00
INV-702,1,2026-07-11,Delhi,BK-103,3,1950.00
INV-702,1,2026-07-11,Delhi,BK-103,3,1950.00
INV-703,1,2026-07-19,Chennai,BK-101,10,4500.00
INV-704,1,2026-07-28,Pune,BK-104,1,350.00

invoice,line,sold_on,city,sku,qty,amount
INV-801,1,2026-08-02,Delhi,BK-102,5,1495.00
INV-802,1,2026-08-09,Chennai,BK-103,2,1300.00
INV-803,1,2026-08-3O,Pune,BK-101,1,450.00
INV-804,1,2026-08-17,Pune,BK-104,4,1400.00
INV-805,1,2026-08-25,Delhi,BK-101,2,900.00

# Raw CSV exports -> a staging table -> clean tables in a database file -> Parquet partitioned by month,
# with the row counts of every stage written down and reconciled.
import os

import duckdb

TYPES = {"invoice": "VARCHAR", "line": "INTEGER", "sold_on": "DATE", "city": "VARCHAR", "sku": "VARCHAR", "qty": "INTEGER", "amount": "DECIMAL(10,2)"}

with duckdb.connect("analytics.duckdb") as db:
    # 1. Stage: every file, with its name, and every line that fails kept aside. Stating the date format makes the
    #    reader strict: without it, the typo 2026-08-3O would load as 2026-08-03.
    db.execute(f"""
        CREATE OR REPLACE TABLE staged AS
        SELECT *, filename AS source
        FROM read_csv('raw/*.csv', header = true, types = {TYPES}, dateformat = '%Y-%m-%d', store_rejects = true)
    """)
    rejected = db.sql("SELECT count(*) FROM reject_errors").fetchone()[0]
    print("set aside:", db.sql("SELECT line, column_name, csv_line FROM reject_errors").fetchall())

    # 2. Clean: one row per invoice line, refunds in a table of their own.
    db.sql("""
        CREATE OR REPLACE TABLE sales AS
        SELECT DISTINCT ON (invoice, line) * EXCLUDE (source) FROM staged WHERE amount > 0 ORDER BY invoice, line
    """)
    db.sql("CREATE OR REPLACE TABLE refunds AS SELECT * EXCLUDE (source) FROM staged WHERE amount < 0")

    # 3. Publish: one folder per month, readable by any tool that reads Parquet.
    db.sql("""
        COPY (SELECT *, date_trunc('month', sold_on)::DATE AS month FROM sales ORDER BY sold_on, invoice, line)
        TO 'sales_by_month' (FORMAT parquet, PARTITION_BY (month), OVERWRITE)
    """)

    lines_in_files = sum(sum(1 for _ in open(os.path.join("raw", f))) - 1 for f in os.listdir("raw"))
    staged, sales, refunds = (db.sql(f"SELECT count(*) FROM {t}").fetchone()[0] for t in ("staged", "sales", "refunds"))
    duplicates = staged - sales - refunds
    print(f"lines in the files {lines_in_files} = staged {staged} + rejected {rejected}")
    print(f"staged {staged} = clean sales {sales} + refunds {refunds} + duplicate lines {duplicates}")
    assert lines_in_files == staged + rejected

published = duckdb.sql("SELECT month, count(*) AS lines, sum(amount) AS amount FROM read_parquet('sales_by_month/*/*.parquet') GROUP BY month ORDER BY month")
print(published)
print("files:", sorted(os.path.join(os.path.basename(d), f) for d, _, fs in os.walk("sales_by_month") for f in fs))

set aside: [(4, 'sold_on', 'INV-803,1,2026-08-3O,Pune,BK-101,1,450.00')]
lines in the files 16 = staged 15 + rejected 1
staged 15 = clean sales 13 + refunds 1 + duplicate lines 1
┌────────────┬───────┬───────────────┐
│   month    │ lines │    amount     │
│    date    │ int64 │ decimal(38,2) │
├────────────┼───────┼───────────────┤
│ 2026-07-01 │     5 │       7999.00 │
│ 2026-08-01 │     4 │       5095.00 │
│ 2026-09-01 │     4 │       4947.00 │
└────────────┴───────┴───────────────┘

files: ['month=2026-07-01/data_0.parquet', 'month=2026-08-01/data_0.parquet', 'month=2026-09-01/data_0.parquet']

The staging stepstaging table: A working table that retains input rows for typed reading, auditing and subsequent transformation before publication. reads every file with stated types and a stated date format, keeps each row’s file name, and sets aside the line it cannot read. The date format matters more than it looks: without it, DuckDB’s cast reads 2026-08-3O as 2026-08-03 and the typo becomes a plausible wrong date, while dateformat = '%Y-%m-%d' makes the reader strict, so the line goes to reject_errors with its line number. The cleaning step keeps one row per invoice line and moves the refund to a table of its own. The publishing step writes one Parquet folder per month.

The two printed equations are the pipeline’s real output. Sixteen lines in the files are 15 staged plus 1 rejected, and 15 staged are 13 clean sales, 1 refund and 1 duplicate. An idempotent outputidempotent output: Output whose declared observable result remains the same when the same operation is repeated with the same input. keeps its stated result when the same input is processed again. A pipeline that cannot write these lines down cannot say whether it lost or invented data.

Use the 16 = 15 + 1 and 15 = 13 + 1 + 1 equations to account for each original export line.

Keep working tables separate from published files

A reader must support both the Parquet data types and the dataset’s partition conventions.

The .duckdb file is the pipeline’s workspace: tables keep their types, queries on them are fast, and the next run can compare against them. The Parquet folders are its output: compatible Parquet dataset readers can read them, and their folder names are the partition values, month=2026-07-01. Only one process may write to a DuckDB file at a time, while several may read it with read_only=True (concurrency), which is a reason to publish Parquet for readers rather than share the working database.

Follow the writable analytics.duckdb tables to three published month folders and their reader boundary.

Choose storage by update and read requirements

Start with what changes and what a reader needs. The database table supports updates to working rows, while a Parquet hand-off is usually replaced as a file or dataset version. Partitioning helps queries that select its keys; it does not automatically reduce every scan. CSV can be appropriate for a spreadsheet export even though it requires explicit parsing and type choices. Check reader support for the stored types and folder conventions before choosing a hand-off format. A count of columns projected by a query describes requested output, not a universal prediction of total bytes or time.

Compare mutable working tables, selected Parquet columns and a CSV reader’s text interpretation.

Each statement names one pipeline operation

• CREATE OR REPLACE TABLE … AS SELECT … builds a table from a query and replaces the previous version.
• INSERT INTO t BY NAME SELECT … appends rows, matching columns by name rather than position.
• COPY (query) TO 'file.parquet' (FORMAT parquet) writes a query’s result to a file.
• PARTITION_BY (month) writes one folder per value, such as month=2026-07-01/data_0.parquet.
• OVERWRITE_OR_IGNORE, OVERWRITE and APPEND decide what a second run does with files already there.
• FILENAME_PATTERN names the files inside each folder, with an index {i} or a random {uuid}.
• EXPORT DATABASE writes a whole database as files plus scripts, and IMPORT DATABASE reads it back.
• The relational API builds the same queries from Python method calls such as .filter() and .aggregate().
• EXPLAIN shows the plan a query will run, and EXPLAIN ANALYZE runs it and reports what each step did.

Default partitioned COPY refuses a nonempty target. APPEND repeats four sales as eight rows. OVERWRITE removes existing output; OVERWRITE_OR_IGNORE replaces matching filenames and may leave others. The twelve-file March fixture reads one file with the month-folder filter and all twelve with the sale-date filter inside the files.

• The Excel extension reads .xlsx files with read_xlsx and writes them with COPY … (FORMAT xlsx).

Identify which operation changes a working table, writes output, describes a plan or exports a database.

Partitioned writes that can be re-run: overwrite modes and reproducible output Pro: opens with a Pro, Premium or Ultimate pass

The relational API: queries built from Python, and the SQL behind them Pro: opens with a Pro, Premium or Ultimate pass

Reading plans with EXPLAIN and EXPLAIN ANALYZE, and moving a database Pro: opens with a Pro, Premium or Ultimate pass

Practise the declared data rules

Interview questions

Key takeaways

• A pipeline stages raw files with stated types, cleans them into tables in a .duckdb file, and publishes Parquet with COPY … TO, partitioned into folders such as month=2026-07-01.
• Stating the date format makes the CSV reader strict; DuckDB’s plain cast read 2026-08-3O as 2026-08-03.
• Write down the row counts of every stage, and make them reconcile: lines in the files, staged, rejected, clean, refunds and duplicates.
• Publish Parquet for readers and keep the database file as the pipeline’s own workspace, because only one process may write to it at a time.