Loading External CSV and Parquet Files

Before you start, make sure these files exist (rerun earlier scripts if needed):

ls data/staging/observations.parquet          # from Day 10: python run_extract.py
ls data/*.csv                                  # from Day 4:  python get_course_data.py
ls data/raw/statcan/18100004.zip               # from Day 10, section 10.3

Then unzip the full CPI table for today's large-file exercises:

mkdir -p data/raw/statcan/18100004
unzip -o data/raw/statcan/18100004.zip -d data/raw/statcan/18100004
ls -lh data/raw/statcan/18100004

13.1 Querying Files Directly (25 min)

This is DuckDB's superpower: a file path can be used like a table, with no import step. Start an in-memory session from the repository root:

duckdb

.mode box
.timer on

Example 1 — Parquet in one line

SELECT * FROM 'data/staging/observations.parquet' LIMIT 5;

Example 2 — Inspect before you trust

DESCRIBE SELECT * FROM 'data/staging/observations.parquet';

SELECT name, source, COUNT(*) AS n, MIN(date) AS first, MAX(date) AS latest
FROM 'data/staging/observations.parquet'
GROUP BY ALL
ORDER BY source, name;

SUMMARIZE SELECT * FROM 'data/staging/observations.parquet';

SUMMARIZE is DuckDB's df.describe(): types, min/max, distinct counts and null percentages for every column.

Example 3 — CSV files

SELECT * FROM 'data/statcan_cpi.csv' LIMIT 5;
DESCRIBE SELECT * FROM 'data/statcan_cpi.csv';

DuckDB sniffs the CSV: it guesses the delimiter, whether there's a header, and each column's type. Check what it decided before building on it.

Example 4 — The explicit reader functions

The short form FROM 'file.csv' calls these functions behind the scenes. Use them directly when you need options:

SELECT * FROM read_csv('data/statcan_cpi.csv', header = true) LIMIT 3;
SELECT * FROM read_parquet('data/staging/observations.parquet') LIMIT 3;

Example 5 — Quoting column names with spaces

StatCan column names contain spaces, so wrap them in double quotes:

SELECT REF_DATE, "Products and product groups" AS component, VALUE
FROM 'data/statcan_cpi.csv'
WHERE "Products and product groups" = 'All-items'
ORDER BY REF_DATE DESC
LIMIT 5;

Quotes in SQL: 'single' for text values, "double" for identifiers (column/table names). Mixing them up is the most common beginner error.

Try it 13.1

Using the staging Parquet file, find the latest value and date of every series whose source is 'boc'. (Hint: arg_max(value, date) from Day 11.)

SELECT name,
       arg_max(value, date) AS latest_value,
       MAX(date)            AS latest_date
FROM 'data/staging/observations.parquet'
WHERE source = 'boc'
GROUP BY name
ORDER BY name;

13.2 Taming Messy CSVs (30 min)

Case 1 — Raw BoC files with text values (from Day 4)

DESCRIBE SELECT * FROM 'data/boc_overnight.csv';

If any row contains text such as Bank holiday, the value column is detected as VARCHAR. Clean it with TRY_CAST, the SQL version of Day 5's pd.to_numeric(errors="coerce"):

SELECT CAST(date AS DATE)          AS date,
       TRY_CAST(value AS DOUBLE)   AS overnight_rate
FROM read_csv('data/boc_overnight.csv', all_varchar = true)
WHERE TRY_CAST(value AS DOUBLE) IS NOT NULL
ORDER BY date DESC
LIMIT 5;

all_varchar = true reads every column as text, so nothing is guessed wrongly. You then convert each column deliberately. For raw layers, this "read everything as text, cast explicitly" approach is the safest pattern.

Case 2 — Telling DuckDB the types yourself

SELECT *
FROM read_csv('data/boc_cad_usd.csv',
              header  = true,
              columns = {'date': 'DATE', 'value': 'VARCHAR'})
LIMIT 5;

Case 3 — FRED's changing headers

The FRED CSV's first column may be DATE or observation_date, and the value column is named after the series. Rename columns by position with a table alias:

SELECT date, TRY_CAST(value AS DOUBLE) AS fed_funds
FROM read_csv('data/fred_fedfunds.csv', all_varchar = true) AS t(date, value)
ORDER BY date DESC
LIMIT 5;

AS t(date, value) renames the columns in order. That's ρ from Day 12, and the same trick as df.columns = ["date", ...] in the notebook's fred_fetch.

Case 4 — Many files at once with glob patterns

Day 10 saved one raw file per fetch under data/raw/fred/<SERIES>/<timestamp>.csv. A glob (*) reads them all, and filename = true records where each row came from:

SELECT filename, COUNT(*) AS n_rows
FROM read_csv('data/raw/fred/*/*.csv', all_varchar = true,
              union_by_name = true, filename = true)
GROUP BY filename
ORDER BY filename;

union_by_name = true lets files with different column names be stacked, like UNION ALL BY NAME from Day 12.

Case 5 — The full StatCan CPI table (1M+ rows)

SELECT COUNT(*) FROM 'data/raw/statcan/18100004/18100004.csv';

DESCRIBE SELECT * FROM 'data/raw/statcan/18100004/18100004.csv';

SELECT CAST(REF_DATE || '-01' AS DATE) AS month,
       GEO                            AS geo,
       "Products and product groups"  AS component,
       VALUE                          AS cpi
FROM 'data/raw/statcan/18100004/18100004.csv'
WHERE GEO IN ('Canada', 'Ontario', 'Quebec', 'British Columbia', 'Alberta')
  AND "Products and product groups" = 'All-items'
  AND REF_DATE >= '2020-01'
ORDER BY month DESC, geo
LIMIT 10;

Note the .timer output: DuckDB scans over a million rows in a fraction of a second.

REF_DATE || '-01' joins text together (|| is SQL's string concatenation), turning '2026-07' into '2026-07-01' so it can be cast to a DATE.

If the first column name looks odd (e.g. "REF_DATE"), the file has a hidden byte-order mark (BOM). That's the same issue the notebook fixes with encoding="utf-8-sig". Recent DuckDB versions handle it automatically. If yours doesn't, run DESCRIBE and copy the exact column name in double quotes.

Speed comparison with pandas (optional, 5 min)

Create day13/speed_test.py:

import time
import duckdb
import pandas as pd

PATH = "data/raw/statcan/18100004/18100004.csv"

t0 = time.perf_counter()
df = pd.read_csv(PATH, low_memory=False, encoding="utf-8-sig")
df.columns = df.columns.str.strip()
out_pd = df[(df["GEO"] == "Canada") &
            (df["Products and product groups"] == "All-items")]
print(f"pandas: {time.perf_counter() - t0:.2f}s, {len(out_pd)} rows")

t0 = time.perf_counter()
out_db = duckdb.sql(f"""
    SELECT * FROM '{PATH}'
    WHERE GEO = 'Canada' AND "Products and product groups" = 'All-items'
""").df()
print(f"duckdb: {time.perf_counter() - t0:.2f}s, {len(out_db)} rows")

python day13/speed_test.py

DuckDB reads only the columns and rows it needs and uses all CPU cores, while pandas loads everything into memory first.

13.3 Parquet: Columnar Storage (23 min)

Row-oriented vs column-oriented

CSV (row by row)                         Parquet (column by column)
2026-06-01,V39079,2.25,boc               date:      2026-06-01, 2026-07-01, ...
2026-07-01,V39079,2.25,boc               series_id: V39079, V39079, ...
...                                      value:     2.25, 2.25, ...
                                         source:    boc, boc, ...

Analytical queries usually touch few columns but many rows (e.g. the average of value). Parquet lets DuckDB read only those columns, and similar values stored together compress extremely well. Parquet also stores the types, so dates stay dates.

Example 1 — Inspecting a Parquet file

SELECT * FROM parquet_schema('data/staging/observations.parquet');
SELECT row_group_id, num_values, compression
FROM parquet_metadata('data/staging/observations.parquet')
LIMIT 5;

Example 2 — Writing Parquet with COPY

Convert the cleaned national CPI to Parquet and compare file sizes:

COPY (
    SELECT CAST(REF_DATE || '-01' AS DATE)  AS month,
           GEO                             AS geo,
           "Products and product groups"   AS component,
           VALUE                           AS cpi
    FROM 'data/raw/statcan/18100004/18100004.csv'
    WHERE VALUE IS NOT NULL
) TO 'data/staging/cpi_all.parquet' (FORMAT parquet);

ls -lh data/raw/statcan/18100004/18100004.csv data/staging/cpi_all.parquet

The Parquet file is typically many times smaller.

Example 3 — Partitioned output

Large datasets are often split into folders by a key column, called partitioning:

COPY (SELECT * FROM 'data/staging/observations.parquet')
TO 'data/staging/observations_by_source'
(FORMAT parquet, PARTITION_BY (source), OVERWRITE_OR_IGNORE);

find data/staging/observations_by_source -name "*.parquet"

You'll see folders like source=boc/, source=fred/. Read them back, and DuckDB turns folder names into a column (hive partitioning):

SELECT source, COUNT(*)
FROM read_parquet('data/staging/observations_by_source/*/*.parquet',
                  hive_partitioning = true)
GROUP BY source;

Filtering on source now skips entire folders without opening them.

13.4 Persisting Data: Schemas, Tables and Views (20 min)

Querying files is great for exploration. For a reliable pipeline we build a warehouse: one DuckDB file with organised, named tables.

Schemas: folders for tables

We'll use a layered structure that mirrors the rest of the course:

Setting up

Create sql/01_schemas.sql:

CREATE SCHEMA IF NOT EXISTS raw;
CREATE SCHEMA IF NOT EXISTS staging;
CREATE SCHEMA IF NOT EXISTS marts;

Create sql/02_raw_tables.sql:

-- Long observations from the Day 10 pipeline
CREATE OR REPLACE TABLE raw.observations AS
SELECT * FROM 'data/staging/observations.parquet';

-- Full StatCan CPI, national and provincial, all components
CREATE OR REPLACE TABLE raw.statcan_cpi AS
SELECT CAST(REF_DATE || '-01' AS DATE)  AS month,
       GEO                             AS geo,
       "Products and product groups"   AS component,
       VALUE                           AS cpi
FROM 'data/raw/statcan/18100004/18100004.csv'
WHERE VALUE IS NOT NULL;

Open the warehouse and run both files:

duckdb macro.duckdb

.mode box
.read sql/01_schemas.sql
.read sql/02_raw_tables.sql
.tables
SELECT COUNT(*) FROM raw.observations;
SELECT COUNT(*) FROM raw.statcan_cpi;

CREATE TABLE ... AS SELECT (often shortened to CTAS) creates and fills a table in one step.

Tables vs views

A table stores data. A view stores a query and runs it every time you select from it.

CREATE OR REPLACE VIEW staging.policy_rates AS
SELECT date, name, value
FROM raw.observations
WHERE name IN ('overnight_rate', 'fed_funds');

SELECT name, COUNT(*), MAX(date) FROM staging.policy_rates GROUP BY name;

dbt calls this choice a materialisation (Days 31–33). You're already making it.

Appending new data with INSERT INTO ... SELECT

INSERT INTO raw.observations
SELECT * FROM 'data/staging/observations.parquet'
WHERE date > (SELECT MAX(date) FROM raw.observations);

This is a simple incremental load: add only rows newer than what you have. Think about what could go wrong here (revisions to old values? a series that lags the others?). dlt handles these cases properly on Days 24–27.

One writer at a time: a DuckDB file can be opened for writing by only one process. If Python reports that the database is locked, .quit your CLI session first.

13.5 DuckDB from Python (18 min)

Everything you've done in the CLI works from Python, and DuckDB and pandas exchange data effortlessly. Create day13/duck_python.py.

Example 1 — Querying and getting a DataFrame back

import duckdb

con = duckdb.connect("macro.duckdb", read_only=True)

df = con.sql("""
    SELECT name, COUNT(*) AS n, MAX(date) AS latest
    FROM raw.observations
    GROUP BY name
    ORDER BY name
""").df()                     # .df() converts the result to pandas

print(df)

Example 2 — Parameters: never paste values into SQL strings

series_name = "overnight_rate"

rates = con.execute(
    "SELECT date, value FROM raw.observations WHERE name = ? ORDER BY date",
    [series_name],
).df()
print(rates.tail())

The ? placeholder is filled safely by DuckDB. Building SQL with f-strings from user input invites errors and SQL injection, a classic security hole.

Example 3 — SQL on a pandas DataFrame

DuckDB can query a DataFrame by its Python variable name:

import pandas as pd

provinces = pd.DataFrame({
    "province": ["BC", "AB", "SK", "ON", "QC"],
    "balance_pct_gdp": [-0.8, 0.2, 0.5, -1.4, -1.2],
})

print(duckdb.sql("""
    SELECT province, balance_pct_gdp
    FROM provinces
    WHERE balance_pct_gdp < 0
    ORDER BY balance_pct_gdp
""").df())

Use whichever tool expresses each step more clearly. Many analysts do heavy filtering and joining in SQL, then modelling and plotting in pandas.

Example 4 — Rebuilding a Day 7 series in SQL

cpi_yoy_input = con.sql("""
    SELECT month, cpi
    FROM raw.statcan_cpi
    WHERE geo = 'Canada' AND component = 'All-items'
    ORDER BY month
""").df().set_index("month")["cpi"]

yoy = cpi_yoy_input.pct_change(12) * 100
print(f"Latest CPI YoY: {yoy.iloc[-1]:.1f}%")
con.close()

Computing the YoY change inside SQL needs window functions. That's the first topic on Day 14.

13.6 Capstone: load_warehouse.py (20 min)

Build a script that anyone can run to recreate the warehouse from scratch, reusing your SQL files.

Requirements:

1. Connect to macro.duckdb (read-write).
2. Run sql/01_schemas.sql and sql/02_raw_tables.sql.
3. Create raw.dim_series from the Day 10 CATALOG (a pandas DataFrame, Example 3 style).
4. Run quality checks in SQL (put them in sql/03_checks.sql or in Python):
   • no duplicate (series_id, date) in raw.observations;
   • no orphan series_id in raw.observations missing from raw.dim_series (anti-join, Day 12);
   • policy rates between −1 and 25.
5. Print a summary table of row counts per table.
6. Exit with a non-zero code if any check fails (Day 10).

"""load_warehouse.py — (re)build macro.duckdb from raw files.

Run from the repository root:  python load_warehouse.py
"""
import sys
from pathlib import Path

import duckdb
import pandas as pd

from macro_extract.catalog import CATALOG

DB = "macro.duckdb"
SQL_DIR = Path("sql")


def run_sql_file(con, path):
    print(f"→ running {path}")
    con.execute(path.read_text())


CHECKS = {
    "duplicate (series_id, date) rows": """
        SELECT COUNT(*) FROM (
            SELECT series_id, date FROM raw.observations
            GROUP BY ALL HAVING COUNT(*) > 1
        )
    """,
    "observations with no matching series": """
        SELECT COUNT(*) FROM raw.observations
        ANTI JOIN raw.dim_series USING (series_id)
    """,
    "policy rates outside [-1, 25]": """
        SELECT COUNT(*) FROM raw.observations
        WHERE name IN ('overnight_rate', 'fed_funds')
          AND (value < -1 OR value > 25)
    """,
}


def main():
    con = duckdb.connect(DB)

    run_sql_file(con, SQL_DIR / "01_schemas.sql")
    run_sql_file(con, SQL_DIR / "02_raw_tables.sql")

    dim_series = pd.DataFrame(CATALOG, columns=["source", "series_id", "name"])
    con.execute("CREATE OR REPLACE TABLE raw.dim_series AS SELECT * FROM dim_series")
    print(f"→ raw.dim_series: {len(dim_series)} series")

    failures = 0
    print("\nQuality checks:")
    for label, sql in CHECKS.items():
        n = con.execute(sql).fetchone()[0]
        status = "✓" if n == 0 else "✗"
        print(f"  {status} {label}: {n}")
        failures += n > 0

    print("\nTable sizes:")
    print(con.sql("""
        SELECT table_schema || '.' || table_name AS table_name,
               estimated_size                   AS approx_rows
        FROM duckdb_tables()
        ORDER BY table_name
    """).df().to_string(index=False))

    con.close()
    if failures:
        print(f"\n{failures} check(s) failed.")
        sys.exit(1)
    print("\n✓ Warehouse built successfully.")


if __name__ == "__main__":
    main()

Run it (close any open CLI session on macro.duckdb first):

python load_warehouse.py

Then explore the result in the CLI:

duckdb macro.duckdb -readonly

.tables
SELECT * FROM raw.dim_series;
SELECT o.name, d.source, COUNT(*) AS n
FROM raw.observations AS o JOIN raw.dim_series AS d USING (series_id)
GROUP BY ALL;

Stretch goals

• Add raw.statcan_gdp from data/statcan_gdp.csv to 02_raw_tables.sql.
• Add a check that every monthly series has no missing months in 2018–2024. (Hint: generate the expected months with SELECT * FROM range(DATE '2018-01-01', DATE '2025-01-01', INTERVAL 1 MONTH) and anti-join.)
• Chain the whole pipeline: python run_extract.py && python load_warehouse.py. The && runs the second command only if the first succeeds, which is why exit codes matter.

Recap: the pipeline so far

APIs (BoC, FRED, StatCan)
   │  python run_extract.py          ← Days 8–10
   ▼
data/raw/…  +  data/staging/observations.parquet
   │  python load_warehouse.py       ← Day 13
   ▼
macro.duckdb   raw.*  (staging.*, marts.* coming soon)

Looking ahead: Days 14–17

Next you'll compute YoY inflation, moving averages and running peaks inside SQL with window functions:

SELECT month, cpi,
       100 * (cpi / LAG(cpi, 12) OVER (ORDER BY month) - 1) AS yoy
FROM raw.statcan_cpi
WHERE geo = 'Canada' AND component = 'All-items';

You'll also structure longer queries with Common Table Expressions (WITH ...), the SQL counterpart of breaking code into named steps.

Day 13 checklist

• I can query CSV and Parquet files directly and inspect them with DESCRIBE and SUMMARIZE
• I clean messy CSVs with all_varchar, explicit column types, TRY_CAST and positional renaming
• I can read many files at once with glob patterns
• I understand why Parquet is faster and smaller, and can write partitioned Parquet
• I organise a warehouse into raw, staging and marts schemas and choose between tables and views
• I can use DuckDB from Python with parameters and pass DataFrames both ways
• One command rebuilds and validates my warehouse

git add .
git status            # macro.duckdb and data/ should NOT be listed
git commit -m "Day 13: loading CSV/Parquet, warehouse build script"
git push