Weather Analytics Pipeline with dbt

Build an ETL pipeline that extracts live weather data and transforms it with dbt.

Introduction

30 Second Summary

Every weather app you check depends on a data pipeline that pulls raw forecasts, cleans them, and reshapes them into answers people can actually use. Building one of these pipelines from scratch is the core skill that separates data engineers from people who just write scripts.

In this project, you will build a complete weather analytics pipeline that extracts live forecast data from an API, loads it into an analytical database, and transforms it through layered models. You will use Python, DuckDB, and dbt to go from raw API data to tested, documented analytical tables.

What You'll Build

You run a single script, watch 7-day weather forecasts for 5 cities land in your database, then run dbt build to transform everything into tested tables you can query for insights like which city has the highest average temperature this week.

By the end of this project, you'll have:

  • A live extraction script that pulls hourly forecasts for 5 cities from the Open-Meteo API into your local DuckDB database.
  • A dbt project where you run dbt build and watch raw data transform into clean, tested analytical tables through staging and mart layers.
  • Auto-generated documentation showing your complete data lineage from source to marts, ready to browse and share as a portfolio piece.
  • Secret Mission: Orchestrate the entire pipeline with Prefect so it runs on a schedule with retry logic and a monitoring dashboard.

Are there any prerequisites?

This project assumes you are comfortable writing Python scripts and have basic SQL knowledge. You do not need prior experience with DuckDB, dbt, or data pipeline architecture.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

You're going to build a complete weather analytics pipeline that extracts live forecast data from a public API, loads it into a local analytical database, and transforms it through production-style layers using industry-standard tools.

Set Up Your Development Environment

Before you can pull weather data or run transformations, you need the right tools installed on your machine.

In this step, you will install Python, set up VS Code as your code editor, create a virtual environment to keep your project dependencies isolated, and install the pipeline libraries you need.

In this step, get ready to:
  • Install Python 3.13 and verify it runs from your terminal.
  • Set up VS Code, a project folder, and a virtual environment.
  • Install dbt-duckdb and requests into your virtual environment.
Install Python 3.13

Python is the language your extraction script and pipeline tools run on. This project requires version 3.13 specifically for compatibility with dbt-duckdb.

  • Open your browser and go to python.org/downloads.
  • Find the latest Python 3.13 release and download the installer for your operating system.

macOS

  • Download the macOS installer (it supports macOS 10.13 and later).
  • Open the downloaded .dmg file and follow the installer prompts.
  • The installer adds Python to your PATH automatically.

Windows

  • Download the Windows installer (64-bit), which is the recommended option.
  • Run the installer. On the first screen, check the box that says Add Python to PATH.
  • Click Install Now and follow the remaining prompts.

Don't skip Add Python to PATH

If you miss the PATH checkbox, your terminal will not recognize python as a command. You would need to reinstall or manually configure your PATH.

Now verify that Python installed correctly. Open a terminal and check the version.

macOS

  • Open Terminal by pressing Cmd+Space, typing Terminal, and pressing Enter.
  • Check your Python version by running:
python3 --version

Windows

  • Open a terminal by pressing the Windows key, typing cmd, and pressing Enter.
  • Check your Python version by running:
python --version

✔️ I see Python 3.13.x

Your terminal prints something like Python 3.13.14. Python is ready to go.

ⓧ I see an older version

If you see a version like 3.9 or 3.10, your system has an older Python that shadows the new install.

  • On macOS, the system Python is older. Try running python3.13 --version to access the version you just installed. If that works, use python3.13 in place of python3 for the rest of this project.
  • On Windows, uninstall the older Python version through Settings then Apps, then re-run the 3.13 installer.
  • Alternatively, visit python.org/downloads and download the 3.13 installer again, making sure to check Add Python to PATH.

ⓧ Command not found

Your terminal does not recognize the Python command. This usually means the installer did not add Python to your PATH.

  • On macOS, re-run the installer from the .dmg file. The macOS installer should set PATH automatically.
  • On Windows, re-run the installer and make sure Add Python to PATH is checked. Then close and reopen your terminal.
  • After reinstalling, close your terminal, open a fresh one, and run the version check again.
Set up VS Code and your project folder

VS Code is a free code editor that gives you a built-in terminal, syntax highlighting, and file navigation in one window.

  • Open your browser and go to code.visualstudio.com.
  • Download the installer for your operating system and run it.
  • Follow the installer prompts to complete installation.

Once installed, open VS Code.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.

Now open a terminal inside VS Code and create your project folder.

  • Open the integrated terminal by pressing Ctrl+` (macOS and Windows).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir weather-pipeline
cd weather-pipeline
  • Confirm the folder was created by running:
pwd

You should see a path ending in weather-pipeline.

Next, create a virtual environment. This keeps your project's packages separate from your system Python.

macOS

  • Create the virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment by running:
source venv/bin/activate

You should see (venv) appear at the start of your terminal prompt.

Windows

  • Create the virtual environment by running:
python -m venv venv
  • Activate the virtual environment by running:
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay local to this project and do not affect other Python projects on your machine.

  • Upgrade pip to the latest version by running:
pip install --upgrade pip

Don't see (venv) in your prompt?

If (venv) is not showing, the virtual environment did not activate. Make sure you ran the activation command for your operating system (source venv/bin/activate on macOS, venv\Scripts\activate on Windows).

On Windows PowerShell, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned and then try the activate command again.

Help me troubleshoot my virtual environment activation

Install pipeline tools

With your virtual environment active, you can now install the libraries your pipeline needs. A single command installs dbt-core, DuckDB, the dbt-duckdb adapter, and the requests library for API calls.

  • Install all pipeline dependencies by running:
pip install dbt-duckdb requests

What does this command install?

The dbt-duckdb package automatically installs dbt-core (the transformation framework) and duckdb (the analytical database) as dependencies. You do not need to install them separately.

The requests library handles HTTP calls to the Open-Meteo weather API.

The installation may take a minute. You will see pip downloading and installing packages in your terminal.

Before you verify, what do you think dbt --version will show if everything installed correctly?

  • Verify dbt installed correctly by running:
dbt --version

You should see output listing dbt-core and dbt-duckdb with their version numbers.

  • Also confirm Python still works inside the virtual environment by running:
python --version

This should print Python 3.13.x. Inside your virtual environment, python points to the correct version regardless of your operating system.

✔️ I see dbt-core and dbt-duckdb versions

Your tools are installed and ready. You have everything you need to start building the pipeline.

ⓧ Command not found or an error

If dbt is not recognized, your virtual environment may not be active. Check that (venv) appears in your terminal prompt.

  • If it does not appear, re-activate your virtual environment (source venv/bin/activate on macOS, venv\Scripts\activate on Windows).
  • Then run pip install dbt-duckdb requests again.
  • If pip itself fails, make sure you are inside the weather-pipeline folder where you created the venv directory.

Your development environment is ready. You have Python, a code editor, and all the pipeline libraries installed. Next up, you will write a script that pulls live weather data from an API and loads it into your database.

Extract Weather Data from the API

Your environment is locked and loaded. Now it is time to give your pipeline something to transform.

Before you can model or aggregate anything, you need real data flowing into your warehouse. In this step, you will build the extraction layer: a Python script that pulls live weather forecasts from the Open-Meteo API and lands them as raw records in DuckDB.

In this step, get ready to:
  • Write a Python script that calls the Open-Meteo API for 5 cities.
  • Parse hourly forecast data and load it into a DuckDB table.
  • Verify that 840 raw records landed in your database.
Create the extraction script

The Open-Meteo API is free, requires no API key, and returns structured JSON with hourly granularity. You will request temperature, precipitation, and wind speed for 7 days across 5 cities.

  • In VS Code, create a new file called extract_load.py in the root of your project folder.
  • Add the imports and city configuration by pasting this code:
import requests
import duckdb
from datetime import datetime

# Cities to pull forecasts for, with their coordinates
CITIES = {
    "London": {"lat": 51.5074, "lon": -0.1278},
    "New York": {"lat": 40.7128, "lon": -74.0060},
    "Tokyo": {"lat": 35.6762, "lon": 139.6503},
    "Sydney": {"lat": -33.8688, "lon": 151.2093},
    "Berlin": {"lat": 52.5200, "lon": 13.4050},
}

# Open-Meteo base endpoint (free, no API key required)
API_URL = "https://api.open-meteo.com/v1/forecast"

What does this code do?

You import three libraries: requests for HTTP calls, duckdb for your analytical database, and datetime to timestamp each extraction run.

The CITIES dictionary maps city names to latitude/longitude pairs. API_URL points at the Open-Meteo forecast endpoint.

  • Below the API_URL line, add the extraction function that loops through each city and collects hourly records:


def extract_weather_data():
    # Collect all hourly records across every city
    all_records = []

    for city_name, coords in CITIES.items():
        # Request 7 days of hourly temp, precipitation, wind
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        response = requests.get(API_URL, params=params)
        response.raise_for_status()
        data = response.json()

        # Flatten the parallel arrays into one record per hour
        hourly = data["hourly"]
        for i in range(len(hourly["time"])):
            all_records.append({
                "city": city_name,
                "latitude": data["latitude"],
                "longitude": data["longitude"],
                "timestamp": hourly["time"][i],
                "temperature_2m": hourly["temperature_2m"][i],
                "precipitation": hourly["precipitation"][i],
                "wind_speed_10m": hourly["wind_speed_10m"][i],
                "extracted_at": datetime.now().isoformat(),
            })

    return all_records

What does this function do?

For each city, it sends a GET request to Open-Meteo asking for 7 days of hourly data. The API returns parallel arrays (one array of timestamps, one of temperatures, one of precipitation, etc.).

The inner loop flattens those parallel arrays into individual records. Each record is one dictionary with the city name, coordinates, a single timestamp, and that hour's weather values.

  • Save extract_load.py.
  • Test the extraction by running this command in your terminal:
python -c "from extract_load import extract_weather_data; records = extract_weather_data(); print(f'Got {len(records)} records')"

You should see Got 840 records printed. That is 5 cities multiplied by 168 hours (7 days of hourly data).

Seeing a connection error?

Make sure you have an active internet connection. The Open-Meteo API requires no key, but your machine must be able to reach api.open-meteo.com.

If you see ModuleNotFoundError: No module named 'requests', confirm your virtual environment is activated (you should see (venv) in your terminal prompt).

help me debug my extract_load.py API call

Add the DuckDB loading function

You have 840 records in memory. Now you need a function that creates a DuckDB table and inserts those records so you can query them with SQL later.

  • Below the extract_weather_data function, add the first half of the loading function that connects to DuckDB and creates the table:


def load_to_duckdb(records):
    # Connect to (or create) the DuckDB file
    con = duckdb.connect("weather_pipeline.duckdb")
    con.execute("CREATE SCHEMA IF NOT EXISTS raw")
    con.execute("DROP TABLE IF EXISTS raw.forecasts")
    # Define the raw table schema
    con.execute("""
        CREATE TABLE raw.forecasts (
            city VARCHAR,
            latitude DOUBLE,
            longitude DOUBLE,
            timestamp VARCHAR,
            temperature_2m DOUBLE,
            precipitation DOUBLE,
            wind_speed_10m DOUBLE,
            extracted_at VARCHAR
        )
    """)

What does this code do?

DuckDB creates the weather_pipeline.duckdb file automatically if it does not already exist. The function creates a raw schema to keep raw data separate from transformed models.

It drops and recreates the raw.forecasts table on each run. This makes the script idempotent. You can re-run it any time to get fresh data without duplicates.

  • Directly below the CREATE TABLE statement (still inside the load_to_duckdb function), add the insert logic and the script's entry point:

    # Bulk insert all records into the raw table
    con.executemany(
        "INSERT INTO raw.forecasts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        [
            (
                r["city"],
                r["latitude"],
                r["longitude"],
                r["timestamp"],
                r["temperature_2m"],
                r["precipitation"],
                r["wind_speed_10m"],
                r["extracted_at"],
            )
            for r in records
        ],
    )

    row_count = con.sql("SELECT count(*) FROM raw.forecasts").fetchone()[0]
    print(f"Loaded {row_count} records into raw.forecasts")
    con.close()


if __name__ == "__main__":
    print("Extracting weather data from Open-Meteo API...")
    records = extract_weather_data()
    print(f"Extracted {len(records)} records for {len(CITIES)} cities")
    load_to_duckdb(records)
    print("Done!")

What does this code do?

The executemany call bulk-inserts every record into the table using parameterized queries. After inserting, it queries the row count and prints it as confirmation.

The if __name__ == "__main__" block ties everything together. It calls extract_weather_data() to pull from the API, then load_to_duckdb() to persist the records.

  • Save extract_load.py.

Seeing an indentation error?

The con.executemany block must be indented inside load_to_duckdb. Make sure it sits at the same indentation level as the con.execute lines above it (one level in from def).

The if __name__ block must be at the top level (zero indentation), not inside any function.

help me fix indentation in my extract_load.py

✔️ Awesome, I've got everything!

Great. Make sure you have saved extract_load.py before running it in the next substep.

ⓧ I'd like to double check the full code

Scroll up and confirm your extract_load.py contains these sections in order:

  • Imports (requests, duckdb, datetime).
  • CITIES dictionary with 5 cities and their coordinates.
  • API_URL set to https://api.open-meteo.com/v1/forecast.
  • extract_weather_data() function that loops through cities and flattens hourly data into records.
  • load_to_duckdb(records) function that creates the schema, table, and inserts records.
  • if __name__ == "__main__" block that calls both functions and prints progress.
Run the pipeline and verify data landed

Before you run this, what do you expect to see? Think about how many records 5 cities with 7 days of hourly data would produce.

  • Run the extraction and loading script:
python extract_load.py

You should see output like this:

Extracting weather data from Open-Meteo API...
Extracted 840 records for 5 cities
Loaded 840 records into raw.forecasts
Done!

Where does 840 come from?

Each city gets 7 days of hourly forecasts. That is 7 x 24 = 168 hours per city. With 5 cities, you get 5 x 168 = 840 total records.

  • Verify the data is actually in DuckDB by querying the row count:
python -c "import duckdb; print(duckdb.connect('weather_pipeline.duckdb').sql('SELECT count(*) FROM raw.forecasts').fetchone())"

You should see (840,) printed. This confirms all 840 records are persisted in your DuckDB file.

Seeing 0 rows or a file not found error?

Make sure you ran python extract_load.py first. The DuckDB file (weather_pipeline.duckdb) is only created when the script runs.

If the script ran but you see fewer than 840 rows, check your internet connection. A failed API call for one city would produce a shorter record count.

help me debug why my DuckDB query shows wrong row count

Your extraction layer is live. You have raw weather data sitting in DuckDB, ready to be transformed. Next up, you will initialize a dbt project and build staging models that clean and type this raw data into an analytics-ready format.

Initialize dbt and Build Staging Models

You have 840 rows of raw weather data sitting in DuckDB. The problem is that this data is messy: column names like temperature_2m are cryptic, timestamps are stored as plain strings, and there is no consistent naming convention.

A staging layer solves this. It applies consistent naming, casts data types, and acts as the single clean source for all downstream models. Without it, every analytical query would need to repeat the same cleaning logic. In this step, you will initialize a dbt project and build a staging model that transforms raw records into clean, typed rows.

In this step, get ready to:
  • Scaffold a dbt project and configure it to connect to your DuckDB file.
  • Create a source definition and a staging model with clean column names and proper types.
  • Run dbt run and confirm your staging model builds successfully.
Scaffold the dbt project

dbt organizes transformation logic into a project folder with a standard structure: models, tests, macros, and configuration files. The dbt init command scaffolds this structure for you.

  • Initialize the dbt project by running this command:
dbt init weather_analytics --skip-profile-setup

What does this command do?

This creates a weather_analytics/ folder containing sample model files, a dbt_project.yml configuration, and standard subdirectories (models, tests, macros, etc.).

The --skip-profile-setup flag tells dbt not to run the interactive connection wizard. You will configure the connection manually in the next step for precise control.

  • Confirm the project folder was created by running:
ls weather_analytics

You should see files including dbt_project.yml and a models/ directory.

  • The scaffold includes example model files you do not need. Delete the example models folder by running:
rm -rf weather_analytics/models/example
  • Now open the weather_analytics/dbt_project.yml file in VS Code.
  • Replace its entire contents with this configuration:
name: 'weather_analytics'
version: '1.0.0'

profile: 'weather_analytics'

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

What does this file do?

The dbt_project.yml tells dbt this directory is a project. The profile key connects this project to a connection configuration (which you will create next). The path keys tell dbt where to find your models, tests, and other resources.

  • Save dbt_project.yml.
  • Now create the connection profile. Create a new file called profiles.yml inside the weather_analytics/ folder.
  • Paste this configuration into profiles.yml:
weather_analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: '../weather_pipeline.duckdb'
      schema: main
      threads: 4

What does this profile do?

The type: duckdb tells dbt to use the DuckDB adapter. The path: '../weather_pipeline.duckdb' points to your existing database file one directory up (since you will run dbt from inside the weather_analytics/ folder).

The schema: main sets the default schema where dbt creates models. The threads: 4 allows dbt to run up to 4 models in parallel.

  • Save profiles.yml.
  • Verify the connection by moving into the project folder and running dbt debug:
cd weather_analytics
dbt debug

You should see a series of green checks confirming your profile loads, your connection works, and your project is valid.

Seeing connection errors?

  • Make sure profiles.yml is inside the weather_analytics/ folder (not in the parent project folder).
  • Check that the path value uses ../ to go up one directory to reach weather_pipeline.duckdb.
  • Confirm the profile name weather_analytics matches the profile: key in dbt_project.yml exactly.

Still stuck? Help me debug my dbt connection error

Create the source definition and staging model

In dbt, a source definition tells dbt about tables that already exist in your database (ones created outside of dbt). Your raw.forecasts table was loaded by your Python script, so dbt needs a source definition to reference it.

  • Create the staging directory for your models by running:
mkdir -p models/staging
  • Create a new file called models/staging/sources.yml and paste this source definition:
version: 2

sources:
  - name: raw
    schema: raw
    description: "Raw weather data extracted from the Open-Meteo API"
    tables:
      - name: forecasts
        description: "Hourly weather forecast records for multiple cities"
        columns:
          - name: city
            description: "City name"
          - name: latitude
            description: "Latitude of the weather grid cell"
          - name: longitude
            description: "Longitude of the weather grid cell"
          - name: timestamp
            description: "ISO 8601 timestamp for the forecast hour"
          - name: temperature_2m
            description: "Air temperature at 2 meters above ground in Celsius"
          - name: precipitation
            description: "Precipitation in millimeters"
          - name: wind_speed_10m
            description: "Wind speed at 10 meters above ground in km/h"
          - name: extracted_at
            description: "ISO 8601 timestamp when the data was extracted"

What is a source definition?

A source definition maps external tables (ones not managed by dbt) into your dbt project. The {{ source('raw', 'forecasts') }} macro in your models will resolve to raw.forecasts in your database.

Adding column descriptions here means your documentation will already know what each field represents before you write any transformation logic.

  • Save sources.yml.
  • Now create the staging model. Create a new file called models/staging/stg_forecasts.sql and paste this SQL:
with source as (
    select * from {{ source('raw', 'forecasts') }}
)

select
    city,
    latitude,
    longitude,
    cast(timestamp as timestamp) as forecast_timestamp,
    temperature_2m as temp_celsius,
    precipitation as precipitation_mm,
    wind_speed_10m as wind_speed_kmh,
    cast(extracted_at as timestamp) as extracted_at
from source

What does this staging model do?

The with source CTE pulls all rows from your raw table via the source macro. The select statement then applies three transformations:

  • Casts string timestamps to proper TIMESTAMP types so you can do date arithmetic downstream.
  • Renames cryptic columns like temperature_2m to readable names like temp_celsius.
  • Adds unit suffixes (_mm, _kmh) so anyone reading the mart models knows the units without checking docs.
  • Save stg_forecasts.sql.
Run dbt and verify the staging model builds

With your source defined and your staging model written, dbt can now compile the SQL and execute it against DuckDB.

  • Before you run this, what do you think dbt will create in DuckDB: a table or a view?
  • Run the models by executing:
dbt run

You should see output that says "Completed successfully" with 1 model created. By default, dbt materializes models as views, so it created a view called main.stg_forecasts in your DuckDB file.

  • Verify the staging data looks correct by querying it. Move back to your project root and run:
cd ..
python -c "import duckdb; print(duckdb.connect('weather_pipeline.duckdb').sql('SELECT * FROM main.stg_forecasts LIMIT 5').fetchdf())"

You should see 5 rows with clean column names: city, latitude, longitude, forecast_timestamp, temp_celsius, precipitation_mm, wind_speed_kmh, and extracted_at. The timestamp columns now show proper datetime values instead of raw strings.

dbt run failed or no rows returned?

  • Make sure you ran dbt run from inside the weather_analytics/ directory (where dbt_project.yml lives).
  • Check that sources.yml uses schema: raw (matching the schema your Python script created).
  • Verify weather_pipeline.duckdb exists in the parent directory by running ls ../weather_pipeline.duckdb.

Need more help? Help me fix my dbt run failure

✔️ Awesome, I've got everything!

Great. Double check you have saved both sources.yml and stg_forecasts.sql in the models/staging/ directory.

ⓧ I'd like to double check the full code

Here are the key files you should have at this point. Scroll up and confirm each one matches.

File 1: weather_analytics/dbt_project.yml

name: 'weather_analytics'
version: '1.0.0'

profile: 'weather_analytics'

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

File 2: weather_analytics/profiles.yml

weather_analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: '../weather_pipeline.duckdb'
      schema: main
      threads: 4

File 3: weather_analytics/models/staging/sources.yml

version: 2

sources:
  - name: raw
    schema: raw
    description: "Raw weather data extracted from the Open-Meteo API"
    tables:
      - name: forecasts
        description: "Hourly weather forecast records for multiple cities"
        columns:
          - name: city
            description: "City name"
          - name: latitude
            description: "Latitude of the weather grid cell"
          - name: longitude
            description: "Longitude of the weather grid cell"
          - name: timestamp
            description: "ISO 8601 timestamp for the forecast hour"
          - name: temperature_2m
            description: "Air temperature at 2 meters above ground in Celsius"
          - name: precipitation
            description: "Precipitation in millimeters"
          - name: wind_speed_10m
            description: "Wind speed at 10 meters above ground in km/h"
          - name: extracted_at
            description: "ISO 8601 timestamp when the data was extracted"

File 4: weather_analytics/models/staging/stg_forecasts.sql

with source as (
    select * from {{ source('raw', 'forecasts') }}
)

select
    city,
    latitude,
    longitude,
    cast(timestamp as timestamp) as forecast_timestamp,
    temperature_2m as temp_celsius,
    precipitation as precipitation_mm,
    wind_speed_10m as wind_speed_kmh,
    cast(extracted_at as timestamp) as extracted_at
from source

Your raw data now has a clean, typed staging layer sitting on top of it. Next up, you will build mart models that aggregate this staging data into daily summaries per city and add data quality tests to catch bad data automatically.

Build Mart Models and Add Data Quality Tests

Your staging model gives you clean, typed hourly weather data. But stakeholders don't want 840 individual rows. They want answers: which city was hottest this week, which day had the most rain, where the wind was strongest.

Mart models aggregate staging data into tables that answer these business questions directly. In this step, you'll also add dbt tests that catch bad data before it reaches consumers. This is the production signal hiring managers look for.

In this step, get ready to:
  • Create a mart model that aggregates hourly data into daily city-level stats.
  • Add data quality tests to your staging model.
  • Run dbt build and verify all models and tests pass.
Create the mart model for daily city weather

Marts sit at the top of your data architecture. They reference staging models and produce the tables that end users query directly.

  • In VS Code, create a new folder called marts inside weather_analytics/models/ by running this command in your terminal:
mkdir models/marts

Why a separate marts folder?

The staging/ folder holds cleaning logic. The marts/ folder holds business-facing aggregations. This separation makes it obvious which models are internal plumbing and which are meant for analysts or dashboards.

  • Create a new file called mart_city_daily_weather.sql inside weather_analytics/models/marts/ and paste in this SQL:
with hourly_data as (
    select * from {{ ref('stg_forecasts') }}
)

select
    city,
    cast(forecast_timestamp as date) as forecast_date,
    round(avg(temp_celsius), 1) as avg_temp_c,
    round(min(temp_celsius), 1) as min_temp_c,
    round(max(temp_celsius), 1) as max_temp_c,
    round(sum(precipitation_mm), 1) as total_precipitation_mm,
    round(max(wind_speed_kmh), 1) as max_wind_speed_kmh,
    count(*) as hourly_records
from hourly_data
group by city, cast(forecast_timestamp as date)

What does this model do?

This model references your staging layer with {{ ref('stg_forecasts') }}. That tells dbt about the dependency so it always builds staging first.

It groups 168 hourly records per city into daily summaries. Each row in the output represents one city on one day, with the average, min, and max temperature, total precipitation, max wind speed, and a count of the hourly records that went into that aggregate.

  • Save mart_city_daily_weather.sql.
  • Run the models to confirm the mart builds successfully:
dbt run

You should see 2 models pass. The output ends with a line like:

Completed successfully

Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2

Seeing an error about stg_forecasts?

Make sure you're running dbt run from inside the weather_analytics folder. If you're in the project root, run cd weather_analytics first.

Also confirm your stg_forecasts.sql file still exists in models/staging/. dbt needs it to build the mart's dependency.

help me debug my dbt run error

Add data quality tests to your staging model

dbt tests are assertions you write in YAML. Every time you run dbt build, dbt checks these assertions against your actual data and fails the build if any assertion is violated.

  • Create a new file called schema.yml inside weather_analytics/models/staging/ and paste in this configuration:
version: 2

models:
  - name: stg_forecasts
    description: "Cleaned and typed hourly weather forecast data"
    columns:
      - name: city
        description: "City name"
        tests:
          - not_null
          - accepted_values:
              values: ['London', 'New York', 'Tokyo', 'Sydney', 'Berlin']
      - name: forecast_timestamp
        description: "Timestamp of the forecast hour"
        tests:
          - not_null
      - name: temp_celsius
        description: "Temperature in Celsius"
      - name: precipitation_mm
        description: "Precipitation in millimeters"
      - name: wind_speed_kmh
        description: "Wind speed in kilometers per hour"

What do these tests check?

  • not_null on city ensures no row is missing a city name. A null city would make downstream aggregations silently wrong.
  • accepted_values on city catches extraction errors. If the API returned a typo like "Londn" or an unexpected city, this test fails before the bad data reaches your mart.
  • not_null on forecast_timestamp ensures every record has a time reference. Without it, you can't group by date in the mart.
  • Save schema.yml.
Run dbt build and verify all tests pass

The dbt build command runs models AND tests in dependency order. It builds staging first, tests it, then builds the mart.

  • Before you run this, predict: will all your tests pass, or will any fail? Think about what data the API actually returned.
  • Run the full build by running:
dbt build

You should see all models build and all tests pass. The final summary looks like:

Done. PASS=5 WARN=0 ERROR=0 SKIP=0 TOTAL=5

Why 5 passes?

dbt counts each model and each test as a separate pass. You have 2 models (stg_forecasts and mart_city_daily_weather) plus 3 tests (not_null on city, accepted_values on city, not_null on forecast_timestamp). That's 5 total.

  • Now query the mart to see your daily aggregates. Run this from the weather_analytics folder:
python -c "import duckdb; print(duckdb.connect('../weather_pipeline.duckdb').sql('SELECT * FROM main.mart_city_daily_weather ORDER BY avg_temp_c DESC LIMIT 5').fetchdf())"

You should see a table with 5 rows showing the hottest city-days in your dataset. Each row has a city, date, average temperature, min/max temps, total precipitation, max wind speed, and a count of hourly records.

Seeing a file not found error on the database?

The path ../weather_pipeline.duckdb assumes you're inside the weather_analytics folder. If you moved to a different directory, adjust the path or run cd weather_analytics first.

help me query my DuckDB mart table

✔️ Awesome, I've got everything!

Great. Make sure both mart_city_daily_weather.sql and schema.yml are saved.

ⓧ I'd like to double check the full code

Here is the mart model. Confirm your file matches exactly:

with hourly_data as (
    select * from {{ ref('stg_forecasts') }}
)

select
    city,
    cast(forecast_timestamp as date) as forecast_date,
    round(avg(temp_celsius), 1) as avg_temp_c,
    round(min(temp_celsius), 1) as min_temp_c,
    round(max(temp_celsius), 1) as max_temp_c,
    round(sum(precipitation_mm), 1) as total_precipitation_mm,
    round(max(wind_speed_kmh), 1) as max_wind_speed_kmh,
    count(*) as hourly_records
from hourly_data
group by city, cast(forecast_timestamp as date)

And here is the staging schema with tests:

version: 2

models:
  - name: stg_forecasts
    description: "Cleaned and typed hourly weather forecast data"
    columns:
      - name: city
        description: "City name"
        tests:
          - not_null
          - accepted_values:
              values: ['London', 'New York', 'Tokyo', 'Sydney', 'Berlin']
      - name: forecast_timestamp
        description: "Timestamp of the forecast hour"
        tests:
          - not_null
      - name: temp_celsius
        description: "Temperature in Celsius"
      - name: precipitation_mm
        description: "Precipitation in millimeters"
      - name: wind_speed_kmh
        description: "Wind speed in kilometers per hour"

Your pipeline now has a proper layered architecture with quality gates built in. Next up, you'll generate documentation that visualises the full lineage from source to mart.

Generate Documentation and Validate the Pipeline

Your pipeline now extracts, loads, and transforms weather data through clean layers. But a pipeline without documentation is a black box to anyone else who encounters it.

Documentation proves you understand data lineage. It makes your project portfolio-ready by showing how data flows from source to final table. In this step, you will document your mart model, generate a browsable documentation site, and then validate the full pipeline end-to-end.

In this step, get ready to:
  • Document your mart model with descriptions and tests in a schema file.
  • Generate and browse auto-documentation showing the full data lineage graph.
  • Run the entire pipeline end-to-end to prove it works as a cohesive unit.
Create a schema file for your mart model

Your staging models already have descriptions and tests from the previous step. The mart model still needs its own documentation file so that dbt can include it in the generated docs site.

  • In VS Code, create a new file called schema.yml inside the weather_analytics/models/marts/ folder.
  • Paste the following YAML into the file:
version: 2

models:
  - name: mart_city_daily_weather
    description: "Daily weather aggregates per city - the main analytical table for weather insights"
    columns:
      - name: city
        description: "City name"
        tests:
          - not_null
      - name: forecast_date
        description: "Date of the forecast"
        tests:
          - not_null
      - name: avg_temp_c
        description: "Average temperature for the day in Celsius"
      - name: min_temp_c
        description: "Minimum temperature for the day in Celsius"
      - name: max_temp_c
        description: "Maximum temperature for the day in Celsius"
      - name: total_precipitation_mm
        description: "Total precipitation for the day in millimeters"
      - name: max_wind_speed_kmh
        description: "Maximum wind speed for the day in km/h"
      - name: hourly_records
        description: "Number of hourly records aggregated"
        tests:
          - not_null

What does this file do?

This YAML file tells dbt what each column in mart_city_daily_weather represents. Each description field appears in the generated documentation site.

The not_null tests on city, forecast_date, and hourly_records ensure these critical columns are never empty. If the extraction script produces incomplete data, these tests will catch it before the table reaches consumers.

  • Save schema.yml.
  • Verify dbt recognizes the new tests by running the following command from within the weather_analytics folder:
dbt build

You should see all models pass and all tests pass (including the new not_null tests on the mart columns).

Seeing a YAML parsing error?

YAML is sensitive to indentation. Make sure each level uses exactly two spaces, not tabs. Check that columns: is indented under the model name, and each - name: entry is indented under columns:.

help me fix a YAML indentation error in my dbt schema.yml file

✔️ Awesome, I've got everything!

Great. Double check you have saved weather_analytics/models/marts/schema.yml.

ⓧ I'd like to double check the full code

Your schema.yml file should match the code block above exactly. Scroll up and confirm the model name is mart_city_daily_weather, all 8 columns are listed with descriptions, and not_null tests appear on city, forecast_date, and hourly_records.

Generate and browse the documentation site

Now that every model has descriptions, you can generate a browsable documentation website that shows your full data lineage.

  • Generate the documentation artifacts by running this command from inside weather_analytics:
dbt docs generate

What does this command do?

This command compiles your project into a manifest.json file (all model definitions, descriptions, and relationships) and queries your database to produce a catalog.json file (actual table metadata like column types). Both files land in the target/ directory.

Together, these files power the documentation website that shows how data flows through your pipeline.

You should see output confirming that the catalog was generated successfully.

  • Launch the documentation site locally by running:
dbt docs serve

What will happen when I run this?

This command starts a local webserver and opens the documentation site in your default browser automatically. The terminal will show the URL where the server is running. Keep the terminal open while you browse the docs.

To stop the server later, press Ctrl+C in the terminal.

  • In the documentation site that opens in your browser, click the lineage graph icon (a connected-nodes icon in the bottom-right corner of the page).
  • You should see the flow: raw.forecasts source leads to stg_forecasts which leads to mart_city_daily_weather.

This lineage graph proves your entire pipeline architecture in one visual. Anyone reviewing your project can see exactly how raw API data transforms into analytical tables.

Documentation site not opening?

If the browser does not open automatically, check the terminal output for the URL (it will look like http://localhost: followed by a port number). Copy that URL and paste it into your browser manually.

If you see an error about a missing catalog, make sure you ran dbt docs generate first.

help me troubleshoot dbt docs serve not opening in my browser

  • Once you have finished exploring the documentation site, switch back to your terminal and press Ctrl+C to stop the docs server.
Run the full pipeline end-to-end

Individual pieces work. Now you need to prove the system works as a cohesive unit by running the full pipeline from extraction through transformation and testing in one go.

  • Navigate back to the project root folder (one level up from weather_analytics) by running:
cd ..

Before you run the pipeline, what do you expect to happen? The extraction script will pull fresh data from the API, overwriting the old forecasts. Then dbt will rebuild all models on top of that fresh data and re-run every test.

  • Run the extraction script to pull fresh weather data:
python extract_load.py

You should see output confirming 840 records were extracted and loaded into raw.forecasts.

  • Now run dbt to transform and test the fresh data by switching into the dbt project folder and running:
cd weather_analytics
dbt build

You should see all models pass and all tests pass with 0 failures. The fresh data has flowed through staging into the mart, and every data quality check has confirmed the results are clean.

Tests failing after re-extraction?

If the accepted_values test on city fails, it means the API returned an unexpected city name. Verify your CITIES dictionary in extract_load.py matches the five cities listed in your test: London, New York, Tokyo, Sydney, Berlin.

help me debug why my dbt tests are failing after running the extraction script again

Your pipeline is fully validated. Data flows from a live API through raw storage, through a typed staging layer, into an aggregated mart. Every layer is tested and documented. Next up, the Secret Mission takes this from a manually triggered pipeline to one with scheduled orchestration.

Secret mission

Orchestrate with Prefect

Your pipeline works, but only when you manually trigger it. In this secret mission, you'll add Prefect orchestration so your pipeline runs on a schedule, retries automatically on failure, and logs every step to a monitoring dashboard.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Project folder on your Desktop (contains extract_load.py, pipeline_flow.py, weather_analytics/, venv/, and weather_pipeline.duckdb).
  • Prefect metadata directory at ~/.prefect/ (created if you completed the Secret Mission).

Keep everything running

No action needed. Choose this if you're still actively building or want to keep your pipeline ready to run again.

  • Your weather_pipeline.duckdb file persists between runs. Each time you run extract_load.py, it replaces the data with a fresh 7-day forecast.
  • The virtual environment stays in your project folder. Activate it with venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux) whenever you return.
  • If you completed the Secret Mission, Prefect's metadata lives in ~/.prefect/ and takes minimal disk space.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files stay on disk ready for next time.

  • If the Prefect server is still running, stop it by pressing Ctrl+C in the terminal where you ran prefect server start.
  • If dbt docs serve is still running, stop it by pressing Ctrl+C in that terminal.
  • Deactivate your virtual environment by running deactivate in your terminal.
  • Your project folder and DuckDB file remain on disk. Reactivate the virtual environment next time you return.

Delete - I don't want to use this again

Remove all project resources and start fresh. Nothing persists outside your machine.

  • If the Prefect server is still running, stop it by pressing Ctrl+C in the terminal where you ran prefect server start.
  • Deactivate your virtual environment by running deactivate in your terminal.
  • Delete your project folder by running the following command (macOS/Linux):
rm -rf ~/Desktop/weather-pipeline
  • On Windows, run this in PowerShell:
Remove-Item -Recurse -Force ~\Desktop\weather-pipeline
  • If you completed the Secret Mission, remove the Prefect metadata directory (macOS/Linux):
rm -rf ~/.prefect
  • On Windows, run this in PowerShell:
Remove-Item -Recurse -Force $env:USERPROFILE\.prefect
  • That removes everything this project created. Nothing remains on your machine.

Nice Work!

Nice Work!

You just built a production-style weather analytics pipeline from scratch. Live data flows from the Open-Meteo API through a layered transformation architecture and lands in tested, documented analytical tables ready for dashboarding.

You've learned how to:

  • Build a Python extraction script that pulls live weather data from the Open-Meteo API for 5 cities and loads it into DuckDB as your analytical warehouse.
  • Create a dbt project with layered architecture (staging models for cleaning, mart models for aggregation) and implement data quality tests that catch bad data before it reaches consumers.
  • Generate auto-documentation showing full data lineage from source through staging to marts, and run the entire pipeline end-to-end as a cohesive system.
  • Secret Mission: Orchestrated the full pipeline with Prefect, adding scheduled execution, automatic retries, and a monitoring dashboard.

Ready to quiz yourself?