Build a Multi-City Weather ETL Pipeline

Build a Python ETL pipeline that loads weather data into a DuckDB star schema.

Introduction

30 Second Summary

Weather data flows into apps, dashboards, and forecasts every second. Behind every weather widget on your phone is a pipeline that pulls raw data, cleans it up, and loads it somewhere useful.

In this project, you will build a Python ETL pipeline that extracts live weather forecasts for 5 cities from a free public API. It transforms the messy data into clean DataFrames and loads it all into a queryable analytics database.

What You'll Build

Picture running a single Python script and watching it pull live weather data from across the globe, clean it up, and answer the question "Which city is the hottest this week?" with a SQL query.

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

A live ETL pipeline that pulls hourly weather data for 5 cities from the Open-Meteo API with zero signup or API keys required.
A clean, analysis-ready DataFrame transformed from messy JSON using pandas, with proper data types, city labels, and null values handled.
A DuckDB analytics database organized as a star schema that you can query with SQL to find the hottest city, the rainiest day, and daily temperature rankings.
Secret Mission: Add a data quality gate that flags missing values, temperature outliers, and stale data before loading into the database.

Are there any prerequisites?

This project assumes you're comfortable with basic Python concepts like variables, loops, and functions. You should also have some familiarity with SQL, especially joins and aggregations.

Before We Start

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

Set Up Your Pipeline Environment

You know what you're building. Before writing any pipeline code, you need a clean workspace with the right tools installed.

Professional data engineers isolate every project's dependencies in its own virtual environment. In this step, you'll create one, install three key libraries, and confirm everything loads correctly.

In this step, get ready to:

Create a project directory and a Python virtual environment.
Install the requests, pandas, and duckdb libraries.
Create the etl.py file and verify all imports work.

Create your project directory and virtual environment

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type VS Code and press Enter to open it.
Open the built-in terminal by pressing Ctrl+` (backtick key, on both macOS and Windows).

Check your Python version by running this command:

python3 --version

✔️ I see version 3.10 or higher

Your Python version is compatible with this project. Move on to the next instruction below.

ⓧ I see an older version

This project requires Python 3.10 or higher. Your current version is too old for the libraries you'll install.

Download and install the latest Python from python.org/downloads.
On macOS, the system Python may still resolve to an older version after installing. If python3 --version still shows the old version, try running python3.12 --version (replacing 3.12 with the version you installed) to confirm the new install exists.
After installing, close and reopen VS Code, then run python3 --version again to confirm.

ⓧ Command not found

Python is not installed yet. On Windows, try running python --version instead. If that also fails, install Python.

Download and install Python 3.10 or higher from python.org/downloads.
On Windows, check the Add Python to PATH checkbox during installation.
After installing, close and reopen VS Code, then run python3 --version again.

Windows users

If python3 is not recognized but python works and shows version 3.10 or higher, use python in place of python3 for all commands in this project.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir weather-etl-pipeline
cd weather-etl-pipeline

Confirm you're in the right folder by running:

pwd

You should see a path ending in weather-etl-pipeline.

Create a Python virtual environment by running this command:

python3 -m venv venv

Confirm the virtual environment folder was created by running:

ls

You should see a venv folder listed.

What is a virtual environment?

A virtual environment is a self-contained Python workspace. Any libraries you install inside it stay isolated from other Python projects on your machine.

This prevents version conflicts between projects. It also makes it easy to share your project's exact dependencies with other people.

Activate the virtual environment by running this command:

macOS

source venv/bin/activate

Linux

source venv/bin/activate

Windows

.\venv\Scripts\Activate.ps1

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, PowerShell may block the activation script. Run this command first to allow it, then try the activation command again:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

On macOS or Linux, make sure you typed source before the path. Without it, the command will not activate the environment.

Still stuck? Help me activate my Python virtual environment.

Install the Python libraries

Your virtual environment is active. Now you can install the three libraries your pipeline needs without affecting anything else on your machine.

Install all three libraries at once by running this command:

pip install requests pandas duckdb

You'll see download progress for each package. The install is complete when you see a line starting with Successfully installed followed by a list of packages.

What are these libraries?

These three libraries map directly to the three stages of an ETL pipeline:

requests makes HTTP calls to web APIs. You'll use it to extract live weather data.
pandas cleans and reshapes tabular data. You'll use it to transform messy JSON into analysis-ready rows and columns.
DuckDB is an analytics database that runs inside your Python script. You'll use it to load and query the cleaned data.

Seeing an installation error?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If it is missing, re-run the activation command from above before trying the install again.

If pip is not found, try pip3 instead.

Still stuck? Help me fix my pip install error.

Create etl.py and verify your imports

The libraries are installed. Now you need a Python file to write your pipeline in, and a quick test to confirm everything loads.

In VS Code, click File in the top menu bar.
Click Open Folder.
Navigate to your Desktop, select the weather-etl-pipeline folder, and click Open.
If VS Code asks you to trust the authors of this folder, click Yes, I trust the authors.

VS Code now shows the weather-etl-pipeline folder in the Explorer sidebar on the left. You should see the venv folder listed.

Open the terminal by pressing Ctrl+` (backtick).
Activate the virtual environment again by running the activation command for your platform:

macOS

source venv/bin/activate

Linux

source venv/bin/activate

Windows

.\venv\Scripts\Activate.ps1

Confirm you see (venv) in your prompt before continuing.

In the Explorer sidebar, hover over the WEATHER-ETL-PIPELINE heading and click the New File icon (a page with a plus sign).
Type etl.py and press Enter.

The file opens in the editor. Now add the import statements that pull in your three installed libraries plus a built-in module you'll need in a later step.

Copy the following code into etl.py:

import requests
import pandas as pd
import duckdb
from datetime import datetime, timezone

print('All libraries loaded')

What does this code do?

import requests loads the HTTP library you'll use to call the weather API.
import pandas as pd loads pandas under the shorthand pd. This is the standard convention in data engineering.
import duckdb loads the analytics database engine.
from datetime import datetime, timezone loads Python's built-in date and time tools. You'll use these later for timestamp handling.
print('All libraries loaded') is a quick sanity check. If this line prints, every import succeeded.

Save etl.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).

Before you run this script, what do you expect to see in the terminal? Think about what those import lines and the print statement should produce.

Run the script by running this command in the terminal:

python3 etl.py

You should see All libraries loaded printed in the terminal. This confirms that all three libraries installed correctly and Python can import them.

✔️ Awesome, I've got everything!

Double check you've saved etl.py. You're ready for the next step.

ⓧ I'd like to double check the full code

Here is the complete etl.py file at this point. Make sure yours matches exactly:

import requests
import pandas as pd
import duckdb
from datetime import datetime, timezone

print('All libraries loaded')

Seeing a ModuleNotFoundError?

This means Python cannot find one of the installed libraries. The most common cause is that your virtual environment is not active.

Check that your terminal prompt shows (venv) at the beginning. If it does not, re-run the activation command.
If the environment is active but the error persists, re-run pip install requests pandas duckdb and try again.

Still stuck? Help me fix my ModuleNotFoundError when running etl.py.

Your environment is ready and all three libraries are working. Next up, you'll use requests to pull live weather data from a real API.

Extract Weather Data from the Open-Meteo API

Your environment is locked in. Now it is time to build the first stage of your ETL pipeline: extraction.

Every data pipeline starts by pulling raw data from a source. You will extract live weather forecasts from the Open-Meteo API for 5 cities around the world. The raw data that comes back is messy, nested, and not ready for analysis. That mess is the whole reason the Transform and Load stages exist.

In this step, get ready to:

Define a list of 5 cities with their geographic coordinates.
Write a function that calls the Open-Meteo API for each city.
Run the script and observe the raw data structure coming back from the API.

Define the city data and API endpoint

The Open-Meteo API needs latitude and longitude coordinates to return weather data for a location. You will define a list of 5 cities with their coordinates, plus the API endpoint to call.

In etl.py, find this line:

print('All libraries loaded')

Delete that entire line. It was a test to confirm your imports worked. You do not need it anymore.
Below your import statements, add the city data and API endpoint by pasting this code:

# --- EXTRACT ---

CITIES = [
    {"name": "London", "latitude": 51.51, "longitude": -0.13},
    {"name": "New York", "latitude": 40.71, "longitude": -74.01},
    {"name": "Tokyo", "latitude": 35.68, "longitude": 139.69},
    {"name": "Sydney", "latitude": -33.87, "longitude": 151.21},
    {"name": "Cairo", "latitude": 30.04, "longitude": 31.24},
]

API_URL = "https://api.open-meteo.com/v1/forecast"

What does this code do?

CITIES is a list of dictionaries, each holding a city's name and its WGS84 latitude/longitude. These coordinates are what the Open-Meteo API requires to look up weather data for a specific location.

API_URL stores the Open-Meteo forecast endpoint. Open-Meteo is completely free for non-commercial use. No API key, no signup, no credit card required.

Write the extraction function and run the pipeline

Now you need a function that loops through each city, calls the API, and collects the raw JSON responses. You will also add a main block that triggers the extraction when you run the script.

Below the API_URL line, add the extraction function and main block by pasting this code:

def extract_weather_data(cities):
    all_data = []
    for city in cities:
        params = {
            "latitude": city["latitude"],
            "longitude": city["longitude"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "timezone": "auto",
            "forecast_days": 7,
        }
        response = requests.get(API_URL, params=params)
        response.raise_for_status()
        data = response.json()
        all_data.append({"city": city, "response": data})
        print(f"Extracted {len(data['hourly']['time'])} hourly records for {city['name']}")
    return all_data


# --- MAIN ---

if __name__ == "__main__":
    print("=== EXTRACT ===\n")
    raw_data = extract_weather_data(CITIES)

What does this code do?

extract_weather_data() loops through each city and sends a GET request to the Open-Meteo API. The params dictionary asks for four hourly weather variables: temperature, humidity, precipitation, and wind speed over a 7-day forecast window.
response.raise_for_status() crashes the script immediately if the API returns an error (like a 404 or 500). This is better than silently collecting bad data.
The if __name__ == "__main__": block is Python's way of saying "only run this code when the script is executed directly." It calls your extraction function and stores the result in raw_data.

What are those hourly parameter names?

The Open-Meteo API uses specific parameter names for each weather variable. temperature_2m means temperature measured 2 meters above ground. relative_humidity_2m is humidity at the same height. wind_speed_10m is wind speed measured 10 meters above ground. These are standard meteorological measurement heights.

Save etl.py.

Before you run this, think about what you expect to see. The function loops through 5 cities and requests 7 days of hourly data. How many hourly data points should each city have?

Run the extraction by running this command:

python3 etl.py

You should see output like this:

=== EXTRACT ===

Extracted 168 hourly records for London
Extracted 168 hourly records for New York
Extracted 168 hourly records for Tokyo
Extracted 168 hourly records for Sydney
Extracted 168 hourly records for Cairo

168 records per city. That is 7 days multiplied by 24 hours. Your script just pulled live weather forecast data for 5 cities from a real API with zero authentication.

What does the raw data actually look like?

The print statements show extraction counts, but the actual data sitting in raw_data is a list of deeply nested JSON objects. Each response contains an hourly object with parallel arrays for time, temperature_2m, relative_humidity_2m, and so on.

Timestamps are plain strings, not proper datetime objects. There are no city labels in the API response itself. Some values may be null if a forecast model could not produce a reading. Loading this directly into a database would create an unreliable, unqueryable mess. That is exactly what the Transform stage will fix in the next step.

Seeing a connection error or timeout?

Make sure you have an active internet connection. The script needs to reach the Open-Meteo API.
If you see a ModuleNotFoundError for requests, your virtual environment may not be activated. Reactivate it and try again.
If you see a SyntaxError, check the f-string on the print line. Make sure the curly braces and quotes match the code above exactly.

Still stuck? Ask for help.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl.py.

ⓧ I'd like to double check the full code

The full file is over 30 lines, so scroll up and confirm each section is present in this order:

Your four import statements at the top (requests, pandas, duckdb, datetime). The print('All libraries loaded') line should be gone.
The # --- EXTRACT --- section with the CITIES list (5 dictionaries) and the API_URL constant.
The extract_weather_data() function that loops through cities, calls the API, and returns all_data.
The # --- MAIN --- section with the if __name__ == "__main__": block that calls extract_weather_data(CITIES).

You have live weather data flowing into your pipeline. Right now it is a pile of nested JSON with string timestamps and no city labels. Next up, you will transform this raw data into clean, analysis-ready DataFrames.

Transform Raw Data into Clean DataFrames

Your extraction step is pulling live weather data from the Open-Meteo API for five cities. But if you look at what came back, the data is nested inside JSON objects, timestamps are plain strings, there are no city labels, and some values could be null.

Loading that mess directly into a database would create unreliable, unqueryable data. In this step, you will build a transformation function that turns the raw JSON into a clean pandas DataFrame with proper types, city labels, and null values handled.

In this step, get ready to:

Parse nested API JSON into a flat DataFrame with analytics-friendly column names.
Handle null values with appropriate defaults so no data is missing.
Print a transformation summary to confirm the data is clean and correctly shaped.

Parse the nested JSON and handle null values

The Open-Meteo API returns an hourly object containing parallel arrays for time, temperature, humidity, precipitation, and wind speed. Your transform function will zip these arrays into rows, attach city labels, and fill any null values.

In etl.py, find the blank line between the end of the extract_weather_data function and the # --- MAIN --- comment.
Copy the following code into that space to create the transform function:

# --- TRANSFORM ---

def transform_weather_data(raw_data):
    frames = []
    for entry in raw_data:
        city = entry["city"]
        hourly = entry["response"]["hourly"]

        df = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(hourly["time"]),
                "temperature_c": hourly["temperature_2m"],
                "humidity_pct": hourly["relative_humidity_2m"],
                "precipitation_mm": hourly["precipitation"],
                "wind_speed_kmh": hourly["wind_speed_10m"],
            }
        )

        df["city_name"] = city["name"]
        df["latitude"] = city["latitude"]
        df["longitude"] = city["longitude"]

        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)

    combined["temperature_c"] = combined["temperature_c"].fillna(combined["temperature_c"].median())
    combined["humidity_pct"] = combined["humidity_pct"].fillna(combined["humidity_pct"].median())
    combined["precipitation_mm"] = combined["precipitation_mm"].fillna(0.0)
    combined["wind_speed_kmh"] = combined["wind_speed_kmh"].fillna(0.0)

What does this code do?

The loop iterates through each city's raw API response. pd.to_datetime converts the ISO timestamp strings into proper datetime objects that can be sorted and grouped.
The API response has no city information attached to its weather data. The three lines adding city_name, latitude, and longitude label every row so you can tell which city each reading belongs to.
pd.concat stacks all five per-city DataFrames into one combined DataFrame.
The fillna calls replace any missing values. Temperature and humidity use the column median so outliers don't skew the fill. Precipitation and wind speed default to 0.0 because a missing reading most likely means no rain or calm wind.

Why median instead of mean for null filling?

The median is resistant to extreme values. If Cairo has one freak 50°C reading, the mean shifts noticeably, but the median stays grounded in the typical range. For weather data with potential outliers, median is a safer default.

Save etl.py.
Confirm the function didn't introduce a syntax error by running:

python3 etl.py

You should still see the extraction output printing 168 hourly records per city. The new function exists but is not called yet, so the output has not changed.

Seeing a syntax error?

Check that the # --- TRANSFORM --- comment and the function are placed between the end of extract_weather_data and # --- MAIN ---. Pasting inside an existing function will break indentation.
Make sure the closing parenthesis of pd.DataFrame(...) is on its own line, aligned with df =.
Help me debug a syntax error in my transform function.

Print a transformation summary and run the full pipeline

The function parses and cleans the data, but it does not yet tell you what the result looks like. Adding print statements at the end lets you verify the shape, types, and null counts every time you run the pipeline.

In etl.py, find the last fillna line inside transform_weather_data. Add the following code below it, still inside the function:

    print(f"\nTransformed DataFrame: {combined.shape[0]} rows, {combined.shape[1]} columns")
    print(f"Columns: {list(combined.columns)}")
    print(f"Dtypes:\n{combined.dtypes}")
    print(f"Null counts:\n{combined.isnull().sum()}")
    print(combined.head())

    return combined

What does this code do?

combined.shape tells you how many rows and columns the DataFrame has. You expect 840 rows (168 hours times 5 cities) and 8 columns.
combined.dtypes confirms timestamps are datetime64 and numeric columns are float64 rather than raw strings.
combined.isnull().sum() should show zero nulls across every column, proving the fillna calls worked.

The function is complete, but the main block does not call it yet. You need to wire it in.

Scroll down to the if __name__ == "__main__": block at the bottom of etl.py. Find this line:

    raw_data = extract_weather_data(CITIES)

Add the following lines directly below it:


    print("\n=== TRANSFORM ===\n")
    clean_df = transform_weather_data(raw_data)

Before you run the script, take a moment to predict: how many rows do you expect the combined DataFrame to have? Each city returns 168 hourly data points across 7 days.

Save etl.py and run the pipeline:

python3 etl.py

You should see the extraction output for all 5 cities, followed by a transform summary showing 840 rows and 8 columns. The null counts should all be 0, and a sample of rows should display with city names, timestamps, and numeric weather readings.

Why 840 rows?

The API returns 168 hourly forecasts per city (24 hours times 7 days). With 5 cities, that gives 168 times 5, which equals 840 rows. If you see a different number, one of your API calls may have returned partial data.

Seeing a NameError or unexpected output?

If you see NameError: name 'transform_weather_data' is not defined, make sure the function definition is above the # --- MAIN --- block, not inside it.
If you see KeyError on a column name, check that the dictionary keys in pd.DataFrame match the API's response fields exactly: temperature_2m, relative_humidity_2m, precipitation, wind_speed_10m.
If you see null counts greater than 0, double-check that all four fillna lines are present and that each one reassigns to the column (e.g. combined["temperature_c"] = combined["temperature_c"].fillna(...)).
Help me fix my transform function output.

✔️ Awesome, I've got everything!

Great. Double-check you have saved etl.py.

ⓧ I'd like to double check the full code

Your etl.py file is over 30 lines now, so here is the checklist of sections in order. Scroll through your file and confirm each one is present.

Imports at the top: requests, pandas, duckdb, and datetime.
# --- EXTRACT --- section: CITIES list, API_URL constant, and the extract_weather_data function.
# --- TRANSFORM --- section: the transform_weather_data function with JSON parsing, null handling, print statements, and return combined.
# --- MAIN --- block: calls extract_weather_data then transform_weather_data, printing section headers for each.

Your data is now clean, typed, and labeled. Next up, you will load this DataFrame into a DuckDB database organized as a star schema, turning your in-memory data into a persistent, queryable analytics database.

Load Data into a DuckDB Star Schema

Your pipeline can now pull live weather data and transform it into a clean DataFrame. The problem is that the data lives in memory. The moment your script finishes, everything disappears.

In this step, you will load the transformed data into a DuckDB database organized as a star schema. This makes the data persistent, queryable, and structured for analytics. The star schema is the dimensional modeling pattern that data engineering interviewers ask about most often.

In this step, get ready to:

Connect to a DuckDB database file and create a dimension table for cities.
Create a fact table for hourly weather readings with a foreign key to the dimension table.
Load the transformed DataFrame into both tables and verify the row counts.

Connect to DuckDB and create the dimension table

A star schema splits data into two kinds of tables. Dimension tables hold descriptive attributes like city names and coordinates. Fact tables hold the measurements (temperature, humidity) and point back to dimensions with foreign keys.

In etl.py, find the # --- MAIN --- comment near the bottom of the file.
Add the following code above that comment to start the load function:

# --- LOAD ---


def load_to_duckdb(df, db_path="weather_analytics.duckdb"):
    # Connect to a DuckDB database file on disk
    con = duckdb.connect(db_path)

    # Drop tables if they exist so the pipeline can rerun cleanly
    con.execute("DROP TABLE IF EXISTS fact_weather")
    con.execute("DROP TABLE IF EXISTS dim_cities")

    # Create the dimension table for city attributes
    con.execute(
        """
        CREATE TABLE dim_cities (
            city_id INTEGER PRIMARY KEY,
            city_name VARCHAR,
            latitude DOUBLE,
            longitude DOUBLE
        )
    """
    )

    # Extract unique cities and assign surrogate keys
    cities = df[["city_name", "latitude", "longitude"]].drop_duplicates().reset_index(drop=True)
    cities["city_id"] = cities.index + 1

    # Populate the dimension table
    con.execute("INSERT INTO dim_cities SELECT city_id, city_name, latitude, longitude FROM cities")

What does this code do?

duckdb.connect(db_path) creates (or opens) a database file called weather_analytics.duckdb on disk. Unlike PostgreSQL, DuckDB needs no running server.
The DROP TABLE IF EXISTS lines let you rerun the pipeline without hitting "table already exists" errors.
dim_cities is the dimension table. It stores each city once, with a surrogate key (city_id) that the fact table will reference.
DuckDB can read a pandas DataFrame directly by variable name. That is why INSERT INTO dim_cities SELECT ... FROM cities works without any manual registration.

What is a star schema?

Picture a star shape. The fact table sits at the center, holding every measurement event (one row per hourly weather reading). Dimension tables surround it like points of a star, each holding descriptive details about one entity (cities, dates, sensors).

This separation avoids repeating "London, 51.51, -0.13" on all 168 hourly rows for London. Instead, the fact table stores a single city_id that points to one row in the dimension table.

Save etl.py. The function is not called from __main__ yet, so the next chunk adds the fact table before you run anything.

Create the fact table and map city IDs

The fact table holds the actual weather measurements. Every row links back to dim_cities through a foreign key, which is how you join facts to dimensions in SQL queries.

In etl.py, add this code directly below the last line you just pasted (the con.execute("INSERT INTO dim_cities ...") line):


    # Create the fact table for hourly weather readings
    con.execute(
        """
        CREATE TABLE fact_weather (
            weather_id INTEGER PRIMARY KEY,
            city_id INTEGER REFERENCES dim_cities(city_id),
            timestamp TIMESTAMP,
            temperature_c DOUBLE,
            humidity_pct DOUBLE,
            precipitation_mm DOUBLE,
            wind_speed_kmh DOUBLE
        )
    """
    )

    # Map each city name to its surrogate key from dim_cities
    city_id_map = dict(zip(cities["city_name"], cities["city_id"]))
    df["city_id"] = df["city_name"].map(city_id_map)
    df["weather_id"] = range(1, len(df) + 1)

What does this code do?

fact_weather stores one row per hourly weather reading. The REFERENCES dim_cities(city_id) clause creates a foreign key that ties each reading to a city in the dimension table.
city_id_map is a dictionary that maps city names like "London" to their numeric surrogate keys (1, 2, 3...). The .map() call applies this mapping across the entire DataFrame in one step.
weather_id gives every row a unique primary key, starting from 1.

Save etl.py.
Confirm the new code has no syntax errors by running:

python3 etl.py

You should see the same extract and transform output as before. The load_to_duckdb function exists now but is not called from __main__ yet, so no load output appears.

Seeing a SyntaxError or IndentationError?

Make sure the new code is indented inside the load_to_duckdb function body (one level in from def).
Check that the triple-quoted SQL strings (""") have matching opening and closing quotes.
Confirm the chunk 2 code sits directly below the con.execute("INSERT INTO dim_cities ...") line, still inside the same function.

Still stuck? Help me fix a SyntaxError in my load_to_duckdb function

Load the data and run the pipeline

The table structure is ready. Now you need to insert the weather readings into fact_weather and print row counts to confirm everything landed correctly.

In etl.py, add this code directly below the df["weather_id"] = range(1, len(df) + 1) line, still inside load_to_duckdb:


    # Select only the columns that match the fact table schema
    fact_df = df[
        [
            "weather_id",
            "city_id",
            "timestamp",
            "temperature_c",
            "humidity_pct",
            "precipitation_mm",
            "wind_speed_kmh",
        ]
    ]

    # Insert all weather readings into the fact table
    con.execute("INSERT INTO fact_weather SELECT * FROM fact_df")

    # Print row counts to confirm the load succeeded
    dim_count = con.execute("SELECT COUNT(*) FROM dim_cities").fetchone()[0]
    fact_count = con.execute("SELECT COUNT(*) FROM fact_weather").fetchone()[0]
    print(f"\nLoaded {dim_count} cities into dim_cities")
    print(f"Loaded {fact_count} weather records into fact_weather")

    return con

What does this code do?

fact_df selects only the seven columns that match the fact_weather schema. This drops columns like city_name that belong in the dimension table, not the fact table.
DuckDB reads fact_df directly from the Python variable. The SELECT * FROM fact_df query treats the pandas DataFrame as if it were a table.
The SELECT COUNT(*) queries verify that data actually landed in both tables. This is your first end-to-end proof that the ETL pipeline works.

Why DuckDB instead of PostgreSQL or SQLite?

DuckDB is purpose-built for analytics. It uses columnar storage and vectorized execution, which makes aggregation queries fast. It installs with a single pip command and needs zero server setup.

PostgreSQL would require installing and running a database server, adding time without teaching ETL concepts. SQLite lacks analytics-oriented features like window functions over large datasets. DuckDB gives you a modern analytics database pattern you will recognize when working with cloud data warehouses like Snowflake or BigQuery.

Now wire up the load step in your pipeline's main block.

Scroll down to the if __name__ == "__main__": block at the bottom of etl.py.
Find this line:

    clean_df = transform_weather_data(raw_data)

Add this code below that line to call the load function, close the connection, and print a completion message:


    print("\n=== LOAD ===")
    con = load_to_duckdb(clean_df)

    con.close()
    print("\nPipeline complete. Database saved to weather_analytics.duckdb")

Save etl.py.

Before you run this, think about what you expect to see. How many cities should appear in dim_cities? How many weather records should land in fact_weather?

Run the full pipeline by executing:

python3 etl.py

At the end of the output, you should see the Load section print Loaded 5 cities into dim_cities and Loaded 840 weather records into fact_weather, followed by Pipeline complete. Database saved to weather_analytics.duckdb.

Confirm the database file now exists on disk by running:

ls weather_analytics.duckdb

You should see weather_analytics.duckdb listed. This file is your persistent analytics database. Even after the script finishes, the data stays on disk.

Not seeing the row counts?

Check that you added the print("\n=== LOAD ===") and con = load_to_duckdb(clean_df) lines inside the if __name__ == "__main__": block, not outside it.
Make sure the return con line is the last line inside load_to_duckdb. If it is missing, the function returns None and con.close() in main will fail.
If you see a NameError about cities, confirm that the chunk 2 code (fact_weather creation) comes after the chunk 1 code (dim_cities creation) inside the same function.

Need more help? Help me debug my load_to_duckdb function

✔️ Awesome, I've got everything!

Great. Double check you have saved etl.py and that your terminal shows 5 cities and 840 weather records.

ⓧ I'd like to double check the full code

Your etl.py file is now over 100 lines. Scroll through and confirm these sections exist in this order:

Imports at the top: requests, pandas, duckdb, datetime.
# --- EXTRACT --- section with CITIES, API_URL, and extract_weather_data().
# --- TRANSFORM --- section with transform_weather_data().
# --- LOAD --- section with load_to_duckdb() containing both CREATE TABLE statements, the ID mapping, the insert, the row-count prints, and return con.
# --- MAIN --- section calling extract_weather_data(), transform_weather_data(), load_to_duckdb(), then con.close().

Your ETL pipeline now extracts, transforms, and loads weather data into a structured analytics database. Next up, you will query this star schema with SQL to surface real weather insights across cities and time periods.

Query Your Weather Analytics Database

Your ETL pipeline now extracts live weather data, transforms it into clean DataFrames, and loads it into a DuckDB star schema. The hard work of building the pipeline is done.

Data sitting in a database without analysis is just storage. In this step, you will write SQL queries against your star schema to surface real weather insights across cities and time periods.

In this step, get ready to:

Find the city with the highest average temperature using a JOIN query.
Identify the rainiest day using GROUP BY and aggregation.
Rank cities by peak temperature each day using a window function.

Find the hottest city with a JOIN query

The star schema you built separates city details (dim_cities) from weather readings (fact_weather). To answer "which city is hottest?", you need to reconnect them with a JOIN.

In etl.py, scroll to the bottom of the file and find the # --- MAIN --- comment.
Add the following run_analytics function above the # --- MAIN --- comment:

# --- ANALYZE ---


def run_analytics(con):
    print("\n--- ANALYTICS ---\n")

    # Query 1: Which city has the highest average temperature?
    print("1. Hottest city (highest average temperature):")
    result = con.execute(
        """
        SELECT d.city_name, ROUND(AVG(f.temperature_c), 1) AS avg_temp_c
        FROM fact_weather f
        JOIN dim_cities d ON f.city_id = d.city_id
        GROUP BY d.city_name
        ORDER BY avg_temp_c DESC
    """
    ).fetchdf()
    print(result.to_string(index=False))

What does this query do?

JOIN dim_cities d ON f.city_id = d.city_id reconnects the fact and dimension tables using the foreign key, so each weather reading gets its city name.
AVG(f.temperature_c) calculates the mean temperature across all hourly readings for each city.
ORDER BY avg_temp_c DESC puts the hottest city at the top of the results.
.fetchdf() returns the query result as a pandas DataFrame for easy printing.

Now you need to call this function from the main block so it runs as part of the pipeline.

In the if __name__ == "__main__": block at the bottom of etl.py, find these two lines:

    con = load_to_duckdb(clean_df)

    con.close()

Add run_analytics(con) between those two lines, so the block now looks like this:

    con = load_to_duckdb(clean_df)

    run_analytics(con)

    con.close()
    print("\nPipeline complete. Database saved to weather_analytics.duckdb")

Save etl.py.
Run the pipeline to see your first analytics query by running this command:

python3 etl.py

After the Extract, Transform, and Load output you already know, you should see a new --- ANALYTICS --- section with all five cities listed by average temperature, hottest first.

Not seeing the analytics output?

Make sure run_analytics(con) appears before con.close(). Calling it after the connection is closed causes an error.
Check that the run_analytics function is defined above the if __name__ block, not inside it.
If you see a NameError, make sure the function name matches exactly: run_analytics (with an underscore).

Still stuck? Ask for help.

Find the rainiest day with aggregation

The first query answered a question about cities. This next query answers a question about time: which day in the forecast has the most total rainfall across all five cities?

In etl.py, find the last line of the run_analytics function (print(result.to_string(index=False))).
Add the following code below it, still inside the run_analytics function:

    # Query 2: Which day has the most total rainfall?
    print("\n2. Rainiest day (most total precipitation):")
    result = con.execute(
        """
        SELECT CAST(f.timestamp AS DATE) AS forecast_date,
               ROUND(SUM(f.precipitation_mm), 1) AS total_precip_mm
        FROM fact_weather f
        GROUP BY forecast_date
        ORDER BY total_precip_mm DESC
        LIMIT 5
    """
    ).fetchdf()
    print(result.to_string(index=False))

What does this query do?

CAST(f.timestamp AS DATE) strips the hour from each timestamp so you can group all 24 hourly readings into a single day.
SUM(f.precipitation_mm) adds up precipitation across all cities and all hours for each day.
LIMIT 5 keeps only the top 5 rainiest days so the output stays readable.

This query does not need a JOIN because precipitation lives directly in fact_weather. You only JOIN when you need columns from the dimension table.

Save etl.py and run the pipeline again:

python3 etl.py

You should now see two analytics results printed. The first shows average temperatures by city. The second shows the top 5 rainiest forecast days with total precipitation in millimeters.

Seeing a syntax error in your SQL?

Check that the new code is indented to match the first query. Every line inside run_analytics must be indented by 4 spaces.
Make sure the triple-quoted SQL string (""") has matching opening and closing quotes.

Still stuck? Ask for help.

Rank cities by peak temperature with a window function

The first two queries used basic aggregation. This final query introduces a window function, which lets you rank rows within groups without collapsing the groups into a single row.

In etl.py, scroll to the last line of the run_analytics function and add the following code below it, still inside the function:

    # Query 3: Rank cities by peak temperature each day
    print("\n3. Daily peak temperature ranking by city:")
    result = con.execute(
        """
        SELECT forecast_date, city_name, max_temp_c, daily_rank
        FROM (
            SELECT CAST(f.timestamp AS DATE) AS forecast_date,
                   d.city_name,
                   ROUND(MAX(f.temperature_c), 1) AS max_temp_c,
                   RANK() OVER (
                       PARTITION BY CAST(f.timestamp AS DATE)
                       ORDER BY MAX(f.temperature_c) DESC
                   ) AS daily_rank
            FROM fact_weather f
            JOIN dim_cities d ON f.city_id = d.city_id
            GROUP BY forecast_date, d.city_name
        )
        WHERE daily_rank <= 3
        ORDER BY forecast_date, daily_rank
    """
    ).fetchdf()
    print(result.to_string(index=False))

What does this query do?

The inner subquery groups by date and city, then calculates each city's peak temperature for that day with MAX(f.temperature_c).
RANK() OVER (PARTITION BY ... ORDER BY ...) assigns a rank within each day's partition. The hottest city each day gets rank 1.
WHERE daily_rank <= 3 in the outer query filters to show only the top 3 cities per day.

Window functions are a common interview topic in data engineering. They let you compute rankings, running totals, and moving averages without losing individual rows.

Save etl.py.

Before you run the pipeline, what do you expect to see in the output? Think about what three analytics sections should appear, and which city you predict will top the daily temperature rankings most often.

Run the full pipeline by running this command:

python3 etl.py

You should see the full Extract, Transform, and Load output followed by three analytics results. The hottest city query shows average temperatures. The rainiest day query lists the top 5 precipitation days. The ranking query shows the top 3 hottest cities for each forecast day.

Why do the numbers change each time?

Your pipeline pulls live forecast data from the Open-Meteo API. Every time you run it, you get the latest 7-day forecast. The specific temperatures, precipitation values, and rankings will change from day to day.

Seeing an error in the ranking query?

The ranking query uses a subquery (a SELECT inside a FROM). Make sure you have matching parentheses around the inner query.
Check that the triple-quoted SQL string has matching """ at the start and end.
Verify that the entire block is indented consistently inside run_analytics.

Still stuck? Ask for help.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl.py and that your terminal shows all three analytics query results.

ⓧ I'd like to double check the full code

Your etl.py is now too long to show in one block. Scroll up through your file and confirm each section is present in this order:

# --- EXTRACT --- section with the CITIES list, API_URL, and extract_weather_data function.
# --- TRANSFORM --- section with the transform_weather_data function.
# --- LOAD --- section with the load_to_duckdb function.
# --- ANALYZE --- section with the run_analytics function containing all three queries.
# --- MAIN --- section with run_analytics(con) called before con.close().

Secret mission

Add a Data Quality Gate

In production, data pipelines break silently. Bad data gets loaded, reports go wrong, and nobody notices until a stakeholder complains. Your mission: add a data quality gate between Transform and Load that catches problems before they reach the database. You will implement completeness, validity, and freshness checks that separate a hobby pipeline from a production-ready one.

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about. Decide whether to keep your files on disk, pause your environment, or delete everything entirely.

Resources you used:

Project directory on your Desktop containing etl.py, the Python virtual environment, and all project files.
weather_analytics.duckdb database file inside your project directory.

Keep everything running

No action needed. Choose this if you plan to keep working on your pipeline or want to run more queries later.

Your weather_analytics.duckdb file stays on disk and takes minimal space. You can query it anytime by reactivating your virtual environment and running python3 etl.py.
The Python libraries (requests, pandas, duckdb) are installed inside the virtual environment and do not affect your system Python.

Pause - I'll come back to this later

Deactivate your virtual environment to free up the terminal session. All your files stay on disk.

Deactivate the virtual environment by running:
Your project directory, etl.py, and weather_analytics.duckdb remain on disk. When you come back, reactivate the virtual environment and pick up where you left off.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and the database. No cloud resources need cleanup.

Deactivate the virtual environment if it is still active:
Navigate to your Desktop and delete the project directory by running these commands. Replace the folder name with the name of the project folder you created in Step 1:

macOS

cd ~/Desktop
rm -rf [[PROJECT_DIR="your project folder name"]]

Windows

cd ~\Desktop
Remove-Item -Recurse -Force [[PROJECT_DIR="your project folder name"]]

This removes etl.py, the weather_analytics.duckdb database, the virtual environment, and all installed packages. Nothing else on your system is affected.

Nice Work!

Nice work! You just built a complete ETL pipeline that pulls live weather data from a public API, transforms it into clean DataFrames, and loads it into a queryable analytics database.

You've learned how to:

Built a Python ETL pipeline that extracts live hourly weather forecast data from the Open-Meteo API for 5 cities using the requests library.
Transformed messy nested JSON into clean, analysis-ready DataFrames using pandas, then loaded them into a DuckDB star schema with dimension and fact tables.
Wrote SQL analytics queries using joins, aggregations, and window functions to surface weather insights across cities and time periods.
Completed an optional Secret Mission to push your data engineering skills further.

Ready to quiz yourself?