Your First dlt Pipelines

21.1 Installing dlt (15 min)

cd /workspaces/macro-data-course
source .venv/bin/activate
pip install "dlt[duckdb]"
pip freeze > requirements.txt
dlt --version

The [duckdb] part installs the extra packages dlt needs for the DuckDB destination.

Add these lines to .gitignore now, before you create any secrets:

.dlt/secrets.toml
*.duckdb
*.duckdb.wal

21.2 The Smallest Possible Pipeline (25 min)

Create day21/first_pipeline.py:

import dlt

# Data: a plain Python list of dictionaries (Day 1 + Day 4!)
decisions = [
    {"decision_date": "2025-01-29", "rate": 3.00, "change_bp": -25},
    {"decision_date": "2025-03-12", "rate": 2.75, "change_bp": -25},
    {"decision_date": "2025-04-16", "rate": 2.25, "change_bp": -50},
]

pipeline = dlt.pipeline(
    pipeline_name="dlt_practice",   # also the name of the DuckDB file
    destination="duckdb",
    dataset_name="practice",        # the schema inside DuckDB
)

load_info = pipeline.run(decisions, table_name="boc_decisions")
print(load_info)

Run it:

python day21/first_pipeline.py
ls *.duckdb

dlt created dlt_practice.duckdb in your current folder. Open it with the tools you already know:

duckdb dlt_practice.duckdb

.mode box
SELECT table_schema, table_name FROM information_schema.tables ORDER BY 1, 2;
SELECT * FROM practice.boc_decisions;
DESCRIBE practice.boc_decisions;

What dlt did for you

1. Created a schema (practice) and a table (boc_decisions).
2. Inferred types: rate → DOUBLE, change_bp → BIGINT, and decision_date → a text or date column depending on how dlt reads the string (next block).
3. Added bookkeeping columns: _dlt_load_id (which run loaded this row) and _dlt_id (a unique row ID).
4. Created system tables:

SELECT * FROM practice._dlt_loads;          -- one row per successful load
SELECT * FROM practice._dlt_version;        -- schema versions

_dlt_loads replaces the fetched_at column you added by hand on Day 10: every row can be traced to the exact run that produced it.

Types: be explicit about dates

Strings that look like dates may stay text. Convert them in Python, and dlt will create a proper DATE column:

from datetime import date

decisions = [
    {"decision_date": date(2025, 1, 29), "rate": 3.00, "change_bp": -25},
    {"decision_date": date(2025, 3, 12), "rate": 2.75, "change_bp": -25},
    {"decision_date": date(2025, 4, 16), "rate": 2.25, "change_bp": -50},
]

Or give dlt a column hint:

load_info = pipeline.run(
    decisions,
    table_name="boc_decisions",
    columns={"decision_date": {"data_type": "date"}},
)

Quit the DuckDB CLI (.quit) before running a pipeline that writes to the same file. DuckDB still allows only one writer at a time (Day 13).

21.3 Running Twice: Write Dispositions (23 min)

Run first_pipeline.py again, then count the rows:

SELECT COUNT(*) FROM practice.boc_decisions;     -- 6, not 3!
SELECT _dlt_load_id, COUNT(*) FROM practice.boc_decisions GROUP BY 1;

By default dlt appends. For a decision log that's a bug: every run duplicates history. The write disposition controls what happens:

load_info = pipeline.run(
    decisions,
    table_name="boc_decisions",
    write_disposition="replace",
)

Run it twice and check that the count stays at 3.

A first look at merge

Macro data gets revised: StatCan updates past months. merge handles that when you tell dlt which columns identify a row:

revised = [
    {"decision_date": date(2025, 4, 16), "rate": 2.25, "change_bp": -50,
     "note": "Tariff shock"},                                   # updated row
    {"decision_date": date(2025, 6, 4), "rate": 2.25, "change_bp": 0,
     "note": "Hold"},                                           # new row
]

pipeline.run(revised, table_name="boc_decisions",
             write_disposition="merge", primary_key="decision_date")

SELECT decision_date, rate, change_bp, note FROM practice.boc_decisions ORDER BY 1;

Notice that the table gained a note column automatically. That's schema evolution, covered in depth on Days 24–27.

Try it 21.3

Load the provinces fiscal table (Day 4) into practice.provinces with replace. Then change Ontario's balance to −1.5 and load again using merge with primary_key="province". Verify there are still 10 rows.

provinces = [
    {"province": "BC", "balance_pct_gdp": -0.8}, {"province": "AB", "balance_pct_gdp": 0.2},
    {"province": "SK", "balance_pct_gdp": 0.5},  {"province": "MB", "balance_pct_gdp": -0.9},
    {"province": "ON", "balance_pct_gdp": -1.4}, {"province": "QC", "balance_pct_gdp": -1.2},
    {"province": "NB", "balance_pct_gdp": -0.5}, {"province": "NS", "balance_pct_gdp": -0.3},
    {"province": "PE", "balance_pct_gdp": -0.7}, {"province": "NL", "balance_pct_gdp": 0.1},
]
pipeline.run(provinces, table_name="provinces", write_disposition="replace")

pipeline.run([{"province": "ON", "balance_pct_gdp": -1.5}],
             table_name="provinces", write_disposition="merge", primary_key="province")

SELECT COUNT(*), MAX(balance_pct_gdp) FILTER (WHERE province = 'ON') FROM practice.provinces;

21.4 Python Generators and yield (25 min)

dlt resources are usually generators, a Python feature you haven't met yet. It's worth learning slowly.

Example 1 — return gives everything at once

def all_months_list(n):
    months = []
    for i in range(n):
        months.append(i)
    return months

print(all_months_list(5))     # [0, 1, 2, 3, 4]

Example 2 — yield gives one item at a time

def months_generator(n):
    for i in range(n):
        print(f"  (producing {i})")
        yield i

gen = months_generator(3)
print(gen)                    # <generator object ...> nothing has run yet!

for m in gen:
    print("received", m)

Output:

  (producing 0)
received 0
  (producing 1)
received 1
  (producing 2)
received 2

The function pauses at each yield and resumes when the next item is requested. It's lazy: work happens only when someone consumes it.

Why this matters for data

Imagine downloading 50 years of daily data for 200 series. With return, everything must fit in memory before anything is saved. With yield, dlt can process and write each batch as it arrives.

def fetch_series_one_by_one(series_ids):
    for sid in series_ids:
        print(f"downloading {sid}...")
        rows = [{"series_id": sid, "value": 1.0}]   # imagine an API call here
        yield rows                                  # hand over one batch, then continue

for batch in fetch_series_one_by_one(["V39079", "FXCADUSD"]):
    print("got", len(batch), "rows")

You can yield single dictionaries or lists of dictionaries (pages). Yielding lists is faster when an API returns many rows at once.

Example 3 — Generators are used up

gen = months_generator(2)
print(list(gen))   # [0, 1]
print(list(gen))   # [] — a generator can only be consumed once

Try it 21.4

Write a generator compounding_path(start, growth, years) that yields a dictionary {"year": t, "level": ...} for each year, reproducing Day 1's compound-growth example (GDP 3180, 1.5%, 5 years). Collect it with list(...).

def compounding_path(start, growth, years):
    level = start
    for t in range(years + 1):
        yield {"year": t, "level": round(level, 1)}
        level *= 1 + growth

print(list(compounding_path(3180, 0.015, 5)))

21.5 @dlt.resource: a Real BoC Resource (23 min)

A resource is a generator decorated with @dlt.resource. The decorator attaches table-level settings: name, write disposition, keys and column hints.

Create day21/boc_resource.py:

from datetime import date

import dlt
from dlt.sources.helpers import requests      # requests with automatic retries

BASE = "https://www.bankofcanada.ca/valet"


@dlt.resource(
    name="boc_observations",
    write_disposition="replace",
    primary_key=("series_id", "date"),
    columns={"value": {"data_type": "double"}},
)
def boc_observations(series_ids, start_date="2006-01-01"):
    """Yield one page of rows per BoC Valet series (long format, as on Day 10)."""
    for sid in series_ids:
        r = requests.get(f"{BASE}/observations/{sid}/json",
                         params={"start_date": start_date}, timeout=30)
        r.raise_for_status()

        rows = []
        for obs in r.json().get("observations", []):
            cell = obs.get(sid)
            if not isinstance(cell, dict):
                continue
            try:
                value = float(cell.get("v"))
            except (TypeError, ValueError):       # '', 'NA', 'Bank holiday', None
                continue
            rows.append({
                "date": date.fromisoformat(obs["d"]),
                "series_id": sid,
                "value": value,
                "source": "boc",
            })
        yield rows


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="dlt_practice",
                            destination="duckdb", dataset_name="practice")
    info = pipeline.run(boc_observations(["V39079", "FXCADUSD"]))
    print(info)

Things to notice:

• The parsing is your Day 8 parse_valet logic. dlt doesn't replace economic knowledge.
• from dlt.sources.helpers import requests is a drop-in replacement for requests that retries automatically on timeouts, 429 and 5xx errors. It replaces your Day 9 make_session().
• primary_key is documentation now and becomes essential for merge later.
• Calling boc_observations([...]) doesn't download anything yet; it creates a resource. The download happens inside pipeline.run, which is the generator's laziness at work.

python day21/boc_resource.py
duckdb dlt_practice.duckdb -c "SELECT series_id, COUNT(*), MAX(date) FROM practice.boc_observations GROUP BY 1"

21.6 Inspecting Pipelines and Loads (13 min)

The load info

print(info)                       # human-readable summary
print(info.has_failed_jobs)       # True/False
print(pipeline.last_trace)        # timings of extract / normalize / load

The dlt command line

dlt pipeline dlt_practice info         # tables, schemas, last run
dlt pipeline dlt_practice trace        # step-by-step timings of the last run

Reading data back into pandas

ds = pipeline.dataset()
df = ds.boc_observations.df()
print(df.groupby("series_id")["value"].describe())

Where dlt keeps its working files

ls ~/.dlt/pipelines/dlt_practice/

This folder holds the pipeline's state, its schema and recent load packages (the extracted and normalized files before loading), much like your Day 9 raw landing zone. If a pipeline gets into a confusing state during practice, you can reset it:

dlt pipeline dlt_practice drop --help     # see options; or simply:
rm -rf ~/.dlt/pipelines/dlt_practice dlt_practice.duckdb

Seeing the inferred schema

print(pipeline.default_schema.to_pretty_yaml())

Find boc_observations in the output. You'll see each column's data_type, nullable, and the primary_key flags you set.

21.7 Mini-Project: A FRED Resource (13 min)

Create day21/fred_resource.py with a resource fred_observations(series_ids, start_date="2006-01-01") that:

1. Downloads the FRED graph CSV for each series (Day 9), using dlt's requests helper.
2. Parses it with Python's built-in csv module (no pandas needed), skipping missing values ("" or ".") and dates before start_date.
3. Yields long rows {date, series_id, value, source="fred"}, one list per series.
4. Uses replace and the same primary key as the BoC resource.

Load FEDFUNDS and DCOILWTICO into the same dlt_practice pipeline and dataset, then query both tables together with UNION ALL BY NAME (Day 12).

import csv
import io
from datetime import date

import dlt
from dlt.sources.helpers import requests

URL = "https://fred.stlouisfed.org/graph/fredgraph.csv"


@dlt.resource(name="fred_observations", write_disposition="replace",
              primary_key=("series_id", "date"),
              columns={"value": {"data_type": "double"}})
def fred_observations(series_ids, start_date="2006-01-01"):
    start = date.fromisoformat(start_date)
    for sid in series_ids:
        r = requests.get(URL, params={"id": sid}, timeout=30)
        r.raise_for_status()
        reader = csv.reader(io.StringIO(r.text))
        next(reader)                                  # skip header (name varies)
        rows = []
        for line in reader:
            if len(line) < 2 or line[1] in ("", "."):
                continue
            d = date.fromisoformat(line[0])
            if d >= start:
                rows.append({"date": d, "series_id": sid,
                             "value": float(line[1]), "source": "fred"})
        yield rows


if __name__ == "__main__":
    pipeline = dlt.pipeline(pipeline_name="dlt_practice",
                            destination="duckdb", dataset_name="practice")
    print(pipeline.run(fred_observations(["FEDFUNDS", "DCOILWTICO"])))

SELECT source, series_id, COUNT(*) AS n, MAX(date) AS latest
FROM (SELECT * FROM practice.boc_observations
      UNION ALL BY NAME
      SELECT * FROM practice.fred_observations)
GROUP BY ALL
ORDER BY source, series_id;

Day 21 checklist

• I can create a pipeline, run it and inspect the result in DuckDB
• I know what _dlt_load_id, _dlt_id and _dlt_loads are for
• I can choose between append, replace and merge
• I understand generators, yield and laziness
• I can write a @dlt.resource with column hints and a primary key
• I can inspect loads with load_info, the dlt CLI and pipeline.dataset()

git add .
git commit -m "Day 21: first dlt pipelines and resources"
git push