Build a Weather Data ETL Pipeline

Extract live weather data from an API, transform it, and load it into DuckDB.

Introduction

30 Second Summary

Every weather app on your phone is powered by data that someone had to collect, clean, and store somewhere useful. The system that handles this is called an ETL pipeline.

In this project, you will build a complete ETL pipeline that pulls live weather data from an API. Your pipeline will clean the data using pandas, then load it into a DuckDB database you can query with SQL.

What You'll Build

A Python-based ETL pipeline that extracts current weather data for multiple cities from the OpenWeatherMap API, transforms it using pandas, and loads it into a DuckDB analytics database.

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

A working extraction script that pulls live weather data from the OpenWeatherMap API for multiple cities.
A transformation layer that cleans raw JSON responses into a structured pandas DataFrame with proper data types and unit conversions.
A DuckDB database file loaded with weather data that you can query with analytical SQL, including window functions.
Secret Mission: Data quality checks that validate your pipeline output and log warnings when data looks suspicious.

Are there any prerequisites?

This project assumes basic familiarity with Python syntax like variables, functions, and loops. No prior SQL or data engineering experience is needed.

Set Up Your Project and Install Dependencies

Every data pipeline starts with an organized workspace. Keeping your code in separate files for each stage of the ETL process makes your project easier to debug, test, and explain to others.

In this step, you'll create a project folder with six Python files, install the key Python packages, and get a free API key from OpenWeatherMap.

In this step, get ready to:

Create a project folder with six Python files for your ETL pipeline.
Install the requests, pandas, and duckdb packages.
Sign up for a free OpenWeatherMap API key and configure it.

Create the project folder and files

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

Click Terminal in VS Code's top menu bar.
Click New Terminal.

A terminal panel opens at the bottom of VS Code. You'll use it to run all the commands in this project.

Move to your Desktop and create the project folder by running these commands:

cd ~/Desktop
mkdir weather-etl
cd weather-etl

Confirm your terminal now shows weather-etl as the current directory.

Create the six Python files your pipeline needs by running:

macOS

touch config.py extract.py transform.py load.py pipeline.py queries.py

Windows

New-Item config.py -ItemType File
New-Item extract.py -ItemType File
New-Item transform.py -ItemType File
New-Item load.py -ItemType File
New-Item pipeline.py -ItemType File
New-Item queries.py -ItemType File

Confirm the files exist by running:

ls

You should see config.py, extract.py, load.py, pipeline.py, queries.py, and transform.py listed in the output.

Don't see all six files?

Make sure you ran the file creation command from inside the weather-etl folder, not your Desktop.
On Windows, if you see an error with the New-Item command, try running each line one at a time.
Help me create the project files for my weather-etl project.

Install the required Python packages

Your pipeline depends on three Python packages. Each one handles a different stage of the ETL process.

Install all three packages by running:

pip install requests==2.33.1 pandas==3.0.2 duckdb==1.5.2

What do these packages do?

Requests sends HTTP requests to web APIs. You'll use it to fetch weather data from OpenWeatherMap.
Pandas is a data analysis library. You'll use it to clean and structure the raw weather data into a tidy table.
DuckDB is an analytics database that runs locally with zero configuration. You'll use it to store and query your weather data with SQL.

Why DuckDB instead of PostgreSQL?

DuckDB installs as a Python package with zero configuration. PostgreSQL requires setting up a separate database server, configuring users, and managing a background service.

DuckDB is designed for analytical queries. You get powerful SQL features like window functions and aggregations without any server setup.

Your terminal should show "Successfully installed" followed by the three package names and their version numbers.

Seeing 'pip not found' or a permission error?

If pip is not recognized, try using pip3 instead.
On macOS or Linux, if you see a permission error, add --user to the end of the install command.
On Windows, try running python -m pip install instead of pip install.
Help me fix my pip installation issue.

Sign up for an OpenWeatherMap API key

Your pipeline needs live weather data. OpenWeatherMap provides a free API that returns current weather conditions for any city in the world.

Open openweathermap.org in your browser.
Click Sign In in the top navigation bar.
Click Create an Account and complete the signup form with your email and a password.
After signing in, click your username in the top navigation bar.
Click My API keys.
Copy the API key shown on the page.

What is an API key?

An API key is a unique string that identifies you when you make requests to an API. OpenWeatherMap uses it to track your usage and enforce rate limits.

The free tier gives you 1,000,000 API calls per month. No credit card required.

New API keys can take up to 10 minutes to activate

If you just created your account, your API key might not work immediately. Wait a few minutes before testing it. You'll know it's active when you get weather data back in the next step.

In VS Code, click on config.py in the Explorer sidebar to open it.
Paste the following code into config.py:

# OpenWeatherMap API configuration
API_KEY = "your-api-key-here"

# List of cities to collect weather data for
CITIES = ["London", "New York", "Tokyo", "Sydney", "Paris"]

# DuckDB database file path
DB_PATH = "weather.duckdb"

Replace your-api-key-here with the API key you copied from OpenWeatherMap. Keep the quotation marks around it.

What does this code do?

API_KEY stores your personal OpenWeatherMap key so your other scripts can import it.
CITIES is the list of cities your pipeline will collect weather data for. You can swap these for any cities you want later.
DB_PATH is the file name for your DuckDB database. All your cleaned weather data gets stored here.

Save config.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
Confirm you see three variables in the file: API_KEY with your personal key, CITIES with five city names, and DB_PATH set to weather.duckdb.

Not sure where your API key is?

Go back to openweathermap.org, sign in, click your username, then My API keys.
Your key is a long string of letters and numbers. Make sure you paste it between the quotation marks in config.py.
Help me find and configure my OpenWeatherMap API key.

✔️ Awesome, I've got everything!

Make sure you've saved config.py before moving on.

ⓧ I'd like to double check the full code

Your config.py should match the structure below. Your API_KEY value will be different since it's your personal key.

# OpenWeatherMap API configuration
API_KEY = "your-api-key-here"

# List of cities to collect weather data for
CITIES = ["London", "New York", "Tokyo", "Sydney", "Paris"]

# DuckDB database file path
DB_PATH = "weather.duckdb"

Your workspace is ready. Next up, you'll write the extraction script that pulls live weather data from the OpenWeatherMap API.

Extract Weather Data from the API

Your project folder is ready and all your packages are installed. Now it's time to write the first stage of your ETL pipeline: the Extract step.

The OpenWeatherMap API serves live weather data for any city in the world. Your job is to write Python code that calls this API, handles the response, and collects raw data for multiple cities.

In this step, get ready to:

Write a function that extracts weather data for a single city from the OpenWeatherMap API.
Write a function that extracts data for multiple cities in a loop.
Test the extraction by running the script and examining the raw JSON response.

Write the extraction function

The OpenWeatherMap Current Weather API has an endpoint that accepts a city name and your API key, then returns weather data as JSON. Your extraction function needs to make an HTTP GET request to this endpoint and handle the response.

Open extract.py in VS Code and add the following code:

import requests
from config import API_KEY, CITIES


def extract_weather(city, api_key):
    """Extract current weather data for a single city from OpenWeatherMap API."""
    # The endpoint for current weather data
    url = "https://api.openweathermap.org/data/2.5/weather"
    # Send city name and API key as query parameters
    params = {
        "q": city,
        "appid": api_key,
    }

    response = requests.get(url, params=params)

    # Check if the request was successful
    if response.status_code == 200:
        print(f"  Successfully extracted data for {city}")
        return response.json()
    else:
        print(f"  Failed to extract data for {city}: {response.status_code}")
        return None

What does this code do?

The first two lines import the requests library for making HTTP calls, plus your API_KEY and CITIES from config.py.

The extract_weather() function builds a request URL with the city name and API key as query parameters. It sends the request using requests.get(), checks if the response status code is 200 (success), and returns the parsed JSON. If the request fails, it prints the error code and returns None.

You won't see a change yet. The next chunk adds a function that uses extract_weather() to pull data for all five cities.

Seeing a red underline on the import?

Make sure you installed requests in the previous step. Run pip install requests again if needed.
Check that config.py exists in the same folder as extract.py. Both files must be in weather-etl.
help me fix an import error in extract.py

Extract data for multiple cities

A single city is useful for testing, but a real pipeline pulls data from multiple sources. This function loops through your list of cities and collects all the responses.

In extract.py, add this function below extract_weather():

def extract_all_cities(cities, api_key):
    """Extract weather data for all cities in the list."""
    raw_data = []
    for city in cities:
        # Call extract_weather for each city
        data = extract_weather(city, api_key)
        if data is not None:
            raw_data.append(data)
    return raw_data

What does this code do?

extract_all_cities() creates an empty list called raw_data, then loops through each city in the list. For every city, it calls extract_weather() and only adds the result if the call was successful (not None). This way, one failed city does not crash the entire pipeline.

You won't see a change yet. The next chunk adds a test block so you can run the script and see real data.

Test the extraction

Your functions are written, but you haven't run them yet. Python's if __name__ == "__main__" pattern lets you add test code that only runs when you execute the file directly.

At the bottom of extract.py, add this test block below extract_all_cities():

if __name__ == "__main__":
    import json

    # Test extraction for a single city
    print("Testing extraction for London...")
    result = extract_weather("London", API_KEY)
    if result:
        print(json.dumps(result, indent=2))

What does this code do?

The if __name__ == "__main__" block only runs when you execute extract.py directly. When other files import functions from this file later, this block gets skipped.

Inside, it calls extract_weather() for London and prints the raw JSON response with nice formatting using json.dumps(). This lets you inspect the exact data the API returns.

Save extract.py.
Run the script by entering this command in your terminal:

python extract.py

You should see Successfully extracted data for London followed by a large block of JSON data printed to your terminal.

Seeing a 401 error?

A 401 status code means your API key was not accepted. New OpenWeatherMap keys can take up to 10 minutes to activate after creation. Wait a few minutes and try again.
Double check that the API_KEY value in config.py matches the key from your OpenWeatherMap account page exactly. Make sure there are no extra spaces.
help me debug my OpenWeatherMap API error

Take a moment to scan the JSON output. This is the raw data your pipeline will transform in the next step.

What fields are in the API response?

The JSON response contains several nested objects with weather data. Here are the key fields you will use in the Transform step:

name is the city name. sys.country is the country code.
main.temp and main.feels_like are temperatures in Kelvin (you will convert these to Celsius later).
main.humidity is the humidity percentage. main.pressure is atmospheric pressure.
wind.speed is the wind speed in meters per second.
weather[0].description is a human-readable weather description like "clear sky" or "light rain".
dt is the data timestamp as a Unix number.

✔️ Awesome, I've got everything!

Double check you have saved extract.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete extract.py file. Compare it against yours to spot any differences.

import requests
from config import API_KEY, CITIES


def extract_weather(city, api_key):
    """Extract current weather data for a single city from OpenWeatherMap API."""
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": api_key,
    }

    response = requests.get(url, params=params)

    if response.status_code == 200:
        print(f"  Successfully extracted data for {city}")
        return response.json()
    else:
        print(f"  Failed to extract data for {city}: {response.status_code}")
        return None


def extract_all_cities(cities, api_key):
    """Extract weather data for all cities in the list."""
    raw_data = []
    for city in cities:
        data = extract_weather(city, api_key)
        if data is not None:
            raw_data.append(data)
    return raw_data


if __name__ == "__main__":
    import json

    print("Testing extraction for London...")
    result = extract_weather("London", API_KEY)
    if result:
        print(json.dumps(result, indent=2))

Your extraction layer is working. It can pull live weather data from the OpenWeatherMap API for any city you add to your list. Next up, you will transform this messy JSON into a clean, structured table using pandas.

Transform the Raw Data with pandas

In the previous step, you extracted raw weather data from the OpenWeatherMap API. The JSON responses contain everything you need, but the data is deeply nested and temperatures are measured in Kelvin.

The "T" in ETL stands for Transform. You'll use pandas to flatten those nested JSON responses into clean rows and columns, convert temperatures to Celsius, and add timestamps.

In this step, get ready to:

Write a function that converts raw JSON responses into a structured pandas DataFrame.
Handle missing values and enforce correct data types.
Test the transformation by running the script and inspecting the output.

Write the transformation function

The OpenWeatherMap API returns temperature in Kelvin and timestamps as Unix integers. Your transformation function needs to extract the useful fields, convert those units, and organize everything into rows and columns.

Open transform.py in VS Code and paste the following code:

import pandas as pd
from datetime import datetime, timezone


def transform_weather_data(raw_data_list):
    """Transform raw API responses into a clean pandas DataFrame."""
    records = []

    for data in raw_data_list:
        record = {
            "city": data["name"],
            "country": data["sys"]["country"],
            "temperature_celsius": round(data["main"]["temp"] - 273.15, 2),
            "feels_like_celsius": round(data["main"]["feels_like"] - 273.15, 2),
            "humidity": data["main"]["humidity"],
            "pressure": data["main"]["pressure"],
            "wind_speed": data["wind"]["speed"],
            "weather_description": data["weather"][0]["description"],
            "data_timestamp": datetime.fromtimestamp(
                data["dt"], tz=timezone.utc
            ).strftime("%Y-%m-%d %H:%M:%S"),
            "pipeline_run_date": datetime.now(timezone.utc).strftime(
                "%Y-%m-%d %H:%M:%S"
            ),
        }
        records.append(record)

    df = pd.DataFrame(records)

    return df

What does this code do?

The function loops through each raw JSON response and pulls out the fields you need into a flat dictionary.
data["main"]["temp"] - 273.15 converts the API's Kelvin temperature to Celsius. Zero Kelvin equals -273.15 degrees Celsius, so subtracting converts the scale.
datetime.fromtimestamp(data["dt"], tz=timezone.utc) converts the Unix timestamp (a big integer like 1714567200) into a human-readable UTC datetime string.
pipeline_run_date records when the pipeline ran. This is useful for tracking data freshness when you accumulate multiple batches.
pd.DataFrame(records) turns the list of dictionaries into a DataFrame. A DataFrame is a table of rows and columns, like a spreadsheet in code.

Save transform.py.

You won't see a change yet. The next substep adds data cleaning, and then you'll test everything together.

Seeing red squiggles in VS Code?

If VS Code underlines the pandas import, make sure you installed the packages in Step 1. Run pip install pandas in your terminal to confirm.
Check that every opening bracket and quote has a matching closing one. The nested dictionary keys are easy to mis-close.

Help me fix import errors in transform.py

Clean and enrich the data

The DataFrame has the right columns, but raw API data can have gaps. A missing weather description or an unexpected null value would cause problems when you load data into the database later.

In transform.py, find the line that says return df inside the transform_weather_data function.
Add the following code directly above return df:

    # Handle any missing values
    df = df.fillna({"weather_description": "unknown", "wind_speed": 0.0})

    # Ensure correct data types
    df["temperature_celsius"] = df["temperature_celsius"].astype(float)
    df["feels_like_celsius"] = df["feels_like_celsius"].astype(float)
    df["humidity"] = df["humidity"].astype(int)
    df["pressure"] = df["pressure"].astype(int)
    df["wind_speed"] = df["wind_speed"].astype(float)

What does this code do?

fillna() replaces any missing values with sensible defaults. If a city's weather description is missing, it becomes "unknown". If wind speed is missing, it defaults to 0.0.
astype() forces each column to the correct type. This catches any unexpected strings or nulls early, before the data reaches your database.

Save transform.py.

You won't see a change yet. The next substep adds a test block so you can run the full transformation.

Test the transformation

Every ETL step needs a test. The if __name__ == "__main__" block lets you run transform.py directly to verify your transformation works before wiring it into the full pipeline.

Add the following test block at the bottom of transform.py, below the transform_weather_data function:

if __name__ == "__main__":
    from extract import extract_all_cities
    from config import API_KEY, CITIES

    print("Testing transformation...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    df = transform_weather_data(raw_data)
    print(df)
    print(f"\nShape: {df.shape}")
    print(f"\nData types:\n{df.dtypes}")

What does this test block do?

It imports your extraction functions and config, then runs the full extract-then-transform flow.
df.shape prints the number of rows and columns, so you can confirm you got data for all 5 cities across 10 columns.
df.dtypes prints each column's data type, confirming your astype() calls worked.

Save transform.py.
Run the transformation test by running this command:

python transform.py

You should see a table printed in your terminal with 5 rows (one per city) and 10 columns. The shape should read (5, 10).

The data types output should show float64 for temperature and wind speed columns, and int64 for humidity and pressure.

Seeing a KeyError or TypeError?

A KeyError means the JSON key path doesn't match. Double check that you have data["main"]["temp"] and data["sys"]["country"] exactly as shown.
A TypeError on the astype() lines usually means a value is None. Make sure the fillna() block appears before the astype() block.
If you see 401 Unauthorized errors during extraction, your API key may still be activating. New keys can take up to 10 minutes.

Help me debug my transform.py output

✔️ Awesome, I've got everything!

Great. Double check you've saved transform.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete transform.py file. Compare it with yours line by line.

import pandas as pd
from datetime import datetime, timezone


def transform_weather_data(raw_data_list):
    """Transform raw API responses into a clean pandas DataFrame."""
    records = []

    for data in raw_data_list:
        record = {
            "city": data["name"],
            "country": data["sys"]["country"],
            "temperature_celsius": round(data["main"]["temp"] - 273.15, 2),
            "feels_like_celsius": round(data["main"]["feels_like"] - 273.15, 2),
            "humidity": data["main"]["humidity"],
            "pressure": data["main"]["pressure"],
            "wind_speed": data["wind"]["speed"],
            "weather_description": data["weather"][0]["description"],
            "data_timestamp": datetime.fromtimestamp(
                data["dt"], tz=timezone.utc
            ).strftime("%Y-%m-%d %H:%M:%S"),
            "pipeline_run_date": datetime.now(timezone.utc).strftime(
                "%Y-%m-%d %H:%M:%S"
            ),
        }
        records.append(record)

    df = pd.DataFrame(records)

    # Handle any missing values
    df = df.fillna({"weather_description": "unknown", "wind_speed": 0.0})

    # Ensure correct data types
    df["temperature_celsius"] = df["temperature_celsius"].astype(float)
    df["feels_like_celsius"] = df["feels_like_celsius"].astype(float)
    df["humidity"] = df["humidity"].astype(int)
    df["pressure"] = df["pressure"].astype(int)
    df["wind_speed"] = df["wind_speed"].astype(float)

    return df


if __name__ == "__main__":
    from extract import extract_all_cities
    from config import API_KEY, CITIES

    print("Testing transformation...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    df = transform_weather_data(raw_data)
    print(df)
    print(f"\nShape: {df.shape}")
    print(f"\nData types:\n{df.dtypes}")

Your raw API data is now a clean, structured DataFrame with readable temperatures and timestamps. Next up, you'll load this data into a DuckDB database so you can query it with SQL.

Load the Data into DuckDB

Your transform function takes messy API responses and produces a clean pandas DataFrame with proper columns and data types.

That DataFrame only lives in memory though. The moment your script finishes, the data disappears. In this step, you'll write the "L" in ETL by loading your cleaned data into DuckDB, a modern analytics database that stores your weather data in a file on disk.

In this step, get ready to:

Write a load function that creates a DuckDB table and inserts data.
Add deduplication logic to prevent duplicate records.
Test the full load process and verify data is stored in the database.

Write the load function

DuckDB stores data in a single file on your computer, similar to SQLite. You connect to it by pointing at a file path, and DuckDB creates the file if it does not already exist.

Your load function needs to create a table with columns that match the DataFrame you built in transform.py. It uses SQL to define the table structure.

In load.py, add the following code:

import duckdb
from config import DB_PATH


def load_to_duckdb(df, db_path):
    """Load transformed DataFrame into DuckDB database."""
    con = duckdb.connect(db_path)

    # Create table if it does not exist
    con.execute("""
        CREATE TABLE IF NOT EXISTS weather_data (
            city VARCHAR,
            country VARCHAR,
            temperature_celsius DOUBLE,
            feels_like_celsius DOUBLE,
            humidity INTEGER,
            pressure INTEGER,
            wind_speed DOUBLE,
            weather_description VARCHAR,
            data_timestamp VARCHAR,
            pipeline_run_date VARCHAR
        )
    """)

What does this code do?

duckdb.connect(db_path) opens (or creates) the database file at the path defined in config.py.
CREATE TABLE IF NOT EXISTS defines the table schema the first time you run the script. On future runs, it skips creation because the table already exists.
Each column type (VARCHAR, DOUBLE, INTEGER) matches the data types in your pandas DataFrame.

You won't see a change yet. The next chunk completes the function with insert logic.

Add a deduplication check

Every production data pipeline needs to handle duplicate data. If you run your pipeline twice and the weather data has not changed, you do not want two identical rows in your database.

Your deduplication logic checks which city-and-timestamp combinations already exist in the table. It only inserts rows that are genuinely new.

In load.py, find the closing """) of the CREATE TABLE block. Add the following code directly below it, still inside the load_to_duckdb function:

    # Check for duplicates before inserting
    existing = con.execute(
        "SELECT city, data_timestamp FROM weather_data"
    ).fetchdf()

    if not existing.empty:
        # Filter out rows that already exist
        merged = df.merge(
            existing, on=["city", "data_timestamp"], how="left", indicator=True
        )
        new_rows = merged[merged["_merge"] == "left_only"].drop(columns=["_merge"])
    else:
        new_rows = df

    if not new_rows.empty:
        con.execute("INSERT INTO weather_data SELECT * FROM new_rows")
        print(f"  Loaded {len(new_rows)} new rows into weather_data")
    else:
        print("  No new rows to load (all duplicates)")

    con.close()

What does this code do?

.fetchdf() runs a SQL query and returns the result as a pandas DataFrame. This grabs every city-and-timestamp pair already stored in the table.
df.merge(..., indicator=True) compares your new data against what already exists. The _merge column flags each row as left_only (new) or both (duplicate).
INSERT INTO weather_data SELECT * FROM new_rows inserts only the genuinely new rows. DuckDB can read a pandas variable directly by name. No manual registration needed.
con.close() closes the database connection and flushes all writes to disk.

Save load.py.

You won't see a change yet. The next substep adds a test block so you can run the function and verify your data.

Why check for duplicates?

Running a pipeline twice with the same data should not double your row count. This property is called idempotency.

Every production data pipeline uses some form of duplicate detection. By checking for existing city-and-timestamp pairs before inserting, your pipeline is safe to re-run at any time without corrupting the database.

Test the load

Your load function is complete. Time to wire up a test that extracts live weather data, transforms it, loads it into DuckDB, then queries the database to prove the data arrived.

At the bottom of load.py, outside the load_to_duckdb function, add the following test block:

if __name__ == "__main__":
    from extract import extract_all_cities
    from transform import transform_weather_data
    from config import API_KEY, CITIES, DB_PATH

    print("Testing load...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    df = transform_weather_data(raw_data)
    load_to_duckdb(df, DB_PATH)

    # Verify the data
    con = duckdb.connect(DB_PATH)
    result = con.execute("SELECT * FROM weather_data").fetchdf()
    print(f"\nRows in database: {len(result)}")
    print(result)
    con.close()

What does this test block do?

It imports the extract and transform functions from the files you built in earlier steps.
It runs the full Extract, Transform, Load sequence for all five cities.
It then opens the database a second time and runs SELECT * FROM weather_data to prove the rows were actually written to disk.

Save load.py.
Run the test by entering this command in your terminal:

python load.py

You should see extraction messages for each city, a line saying Loaded 5 new rows into weather_data, and then a table of your weather data printed to the terminal.

Confirm the database file exists by running:

ls

You should see weather.duckdb listed alongside your Python files in the project folder.

Seeing an error when you run load.py?

ModuleNotFoundError: No module named 'duckdb' means DuckDB is not installed. Run pip install duckdb and try again.
IndentationError means the deduplication code is not indented inside the load_to_duckdb function. Make sure every line from the dedup chunk has 4 spaces of indentation.
If you see Failed to extract data for every city, your API key may not be active yet. New OpenWeatherMap keys can take up to 10 minutes to activate.
Still stuck? Help me debug my load.py error

✔️ Awesome, I've got everything!

Great. Double check you have saved load.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete load.py file. Compare it against yours and fix any differences.

import duckdb
from config import DB_PATH


def load_to_duckdb(df, db_path):
    """Load transformed DataFrame into DuckDB database."""
    con = duckdb.connect(db_path)

    # Create table if it does not exist
    con.execute("""
        CREATE TABLE IF NOT EXISTS weather_data (
            city VARCHAR,
            country VARCHAR,
            temperature_celsius DOUBLE,
            feels_like_celsius DOUBLE,
            humidity INTEGER,
            pressure INTEGER,
            wind_speed DOUBLE,
            weather_description VARCHAR,
            data_timestamp VARCHAR,
            pipeline_run_date VARCHAR
        )
    """)

    # Check for duplicates before inserting
    existing = con.execute(
        "SELECT city, data_timestamp FROM weather_data"
    ).fetchdf()

    if not existing.empty:
        # Filter out rows that already exist
        merged = df.merge(
            existing, on=["city", "data_timestamp"], how="left", indicator=True
        )
        new_rows = merged[merged["_merge"] == "left_only"].drop(columns=["_merge"])
    else:
        new_rows = df

    if not new_rows.empty:
        con.execute("INSERT INTO weather_data SELECT * FROM new_rows")
        print(f"  Loaded {len(new_rows)} new rows into weather_data")
    else:
        print("  No new rows to load (all duplicates)")

    con.close()


if __name__ == "__main__":
    from extract import extract_all_cities
    from transform import transform_weather_data
    from config import API_KEY, CITIES, DB_PATH

    print("Testing load...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    df = transform_weather_data(raw_data)
    load_to_duckdb(df, DB_PATH)

    # Verify the data
    con = duckdb.connect(DB_PATH)
    result = con.execute("SELECT * FROM weather_data").fetchdf()
    print(f"\nRows in database: {len(result)}")
    print(result)
    con.close()

Your weather data is now stored in a real database file. Next up, you'll wire all three steps together into a single pipeline script that runs Extract, Transform, and Load in one command.

Orchestrate the Full Pipeline

You now have three standalone modules: one that extracts weather data, one that transforms it, and one that loads it into DuckDB. Each one works on its own, but running them manually in the right order every time is tedious and error-prone.

In this step, you will write a single pipeline script that calls Extract, Transform, and Load in sequence. This is called orchestration. You will run the pipeline multiple times to accumulate real weather snapshots in your database.

In this step, get ready to:

Write a pipeline script that orchestrates the full ETL flow.
Run the pipeline multiple times to accumulate weather data.
Verify that multiple batches of data are stored in your database.

Write the pipeline script

Every production data pipeline has an orchestrator. It is the single entry point that kicks off each stage in the correct order and reports on the result.

Open pipeline.py in VS Code and add the following code:

from extract import extract_all_cities
from transform import transform_weather_data
from load import load_to_duckdb
from config import API_KEY, CITIES, DB_PATH
import duckdb


def run_pipeline():
    """Run the full ETL pipeline."""
    print("=" * 50)
    print("WEATHER DATA ETL PIPELINE")
    print("=" * 50)

    # Step 1: Extract
    print("\n[1/3] Extracting weather data...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    print(f"  Extracted data for {len(raw_data)} cities")

    # Step 2: Transform
    print("\n[2/3] Transforming data...")
    df = transform_weather_data(raw_data)
    print(f"  Transformed {len(df)} records")
    print(f"  Columns: {list(df.columns)}")

    # Step 3: Load
    print("\n[3/3] Loading data into DuckDB...")
    load_to_duckdb(df, DB_PATH)

    # Summary
    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    con = duckdb.connect(DB_PATH)
    count = con.execute("SELECT COUNT(*) FROM weather_data").fetchone()[0]
    print(f"Total rows in database: {count}")
    con.close()
    print("=" * 50)


if __name__ == "__main__":
    run_pipeline()

What does this code do?

The imports at the top pull in each function you built in the previous steps. This is why you split the project into separate files. Each module is a reusable piece that the orchestrator snaps together.

run_pipeline() calls Extract, Transform, and Load in sequence, printing progress at each stage so you can see exactly where the pipeline is.
After loading, the summary section opens a fresh connection to DuckDB and runs SELECT COUNT(*) to confirm how many total rows are in the database.
The if __name__ == "__main__" block lets you run the pipeline directly from the terminal.

Save pipeline.py.
Run the pipeline by entering this command in your terminal:

python pipeline.py

You should see output with a header, progress for each of the three stages, and a summary line showing the total row count in your database.

Seeing a ModuleNotFoundError?

Make sure your terminal is inside the weather-etl folder. Run ls and confirm you see pipeline.py, extract.py, transform.py, and load.py all listed.
Check that the file names match the imports exactly. For example, from extract import extract_all_cities expects a file called extract.py.
If you see an API error (status code 401), your API key may have expired or been entered incorrectly in config.py.

Still stuck? Ask for help.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file:

from extract import extract_all_cities
from transform import transform_weather_data
from load import load_to_duckdb
from config import API_KEY, CITIES, DB_PATH
import duckdb


def run_pipeline():
    """Run the full ETL pipeline."""
    print("=" * 50)
    print("WEATHER DATA ETL PIPELINE")
    print("=" * 50)

    # Step 1: Extract
    print("\n[1/3] Extracting weather data...")
    raw_data = extract_all_cities(CITIES, API_KEY)
    print(f"  Extracted data for {len(raw_data)} cities")

    # Step 2: Transform
    print("\n[2/3] Transforming data...")
    df = transform_weather_data(raw_data)
    print(f"  Transformed {len(df)} records")
    print(f"  Columns: {list(df.columns)}")

    # Step 3: Load
    print("\n[3/3] Loading data into DuckDB...")
    load_to_duckdb(df, DB_PATH)

    # Summary
    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    con = duckdb.connect(DB_PATH)
    count = con.execute("SELECT COUNT(*) FROM weather_data").fetchone()[0]
    print(f"Total rows in database: {count}")
    con.close()
    print("=" * 50)


if __name__ == "__main__":
    run_pipeline()

Run the pipeline to accumulate data

A single snapshot of weather data is useful, but the real power of a pipeline is running it repeatedly to build up a history. Each run captures what the weather looks like right now across your five cities.

Why run it more than once?

In production, ETL pipelines run on a schedule (hourly, daily, weekly). Each run adds a new batch of data. Accumulating multiple batches is what makes analytics possible. You can spot trends, compare readings, and calculate changes over time.

Wait about 10 to 15 minutes after your first run. The OpenWeatherMap API updates its data roughly every 10 minutes. If you run again too quickly, the timestamps will match your previous run and the deduplication check will filter everything out.
Run the pipeline again:

python pipeline.py

You should see the same three-stage progress output, followed by a Loaded 5 new rows message and an increased total row count.

If you have time, wait another 10 to 15 minutes and run the pipeline a third time to accumulate even more data.

Seeing "No new rows to load (all duplicates)"?

This means the API returned the same timestamps as your previous run. The deduplication logic in load.py is working correctly. Wait a bit longer and try again.

Need help figuring out timing?

Verify the accumulated data

Each time you ran the pipeline, the summary section printed the total number of rows in your database. That count should have grown with each successful run.

Look at the Total rows in database: line from your most recent pipeline run.
Confirm the count is higher than 5. If you started with one batch from the load.py test run and ran the pipeline twice, you should see around 15 rows (three batches of 5 cities).

What is happening inside the database?

Your weather.duckdb file now contains multiple snapshots of weather conditions across five cities. Each batch has a unique data_timestamp from the API and a pipeline_run_date recording when you ran the pipeline.

This accumulated data is what makes the analytical queries in the next step possible. You will be able to compare temperatures across cities and track changes between readings.

Your pipeline is running end to end. You have real weather data stacking up in your database. Next up, you will write SQL queries to analyze the data you have collected.

Query Your Data with SQL

Your ETL pipeline is fully wired up. Each run of pipeline.py extracts fresh weather data, transforms it, and loads it into your DuckDB database.

Now it is time to put that data to work. In this step, you will write SQL queries that slice through your weather data to find patterns, compare cities, and track changes over time.

In this step, get ready to:

Write SQL queries to filter and aggregate weather data.
Use a window function to track temperature changes between readings.
Export a query result to a CSV file.

Write analytical SQL queries

Your weather.duckdb database already contains multiple batches of weather data. A new file called queries.py will hold all of your analytical queries in one place.

Open queries.py in VS Code.
Set up the file with its imports, a run_queries() function, and your first query by copying the following code:

import duckdb
from config import DB_PATH


def run_queries():
    """Run analytical SQL queries on the weather data."""
    con = duckdb.connect(DB_PATH)

    # Query 1: All data for a specific city
    print("=" * 50)
    print("Query 1: All weather data for London")
    print("=" * 50)
    result = con.execute("""
        SELECT city, temperature_celsius, humidity, weather_description, data_timestamp
        FROM weather_data
        WHERE city = 'London'
        ORDER BY data_timestamp DESC
    """).fetchdf()
    print(result)

    con.close()


if __name__ == "__main__":
    run_queries()

What does this code do?

This file connects to your DuckDB database using the same DB_PATH from config.py.

Query 1 uses a WHERE clause to filter down to London's data only. ORDER BY data_timestamp DESC sorts results so the most recent reading appears first.

Save queries.py.
Run the file to test your first query:

python queries.py

You should see a table of London's weather readings printed in your terminal, with the most recent reading at the top.

Seeing an error?

If you see Table with name weather_data does not exist, make sure you ran pipeline.py at least once in the previous step.
If you see an empty table with column headers but no rows, your database has no London data. Run pipeline.py again to load a fresh batch.
Why is my query returning no data?

Filtering by one city is useful, but the real power of SQL is aggregation. The next two queries will summarize your data across all five cities.

In queries.py, find the line that says con.close().
Add the following two queries directly above con.close():

    # Query 2: Average temperature per city
    print("\n" + "=" * 50)
    print("Query 2: Average temperature per city")
    print("=" * 50)
    result = con.execute("""
        SELECT
            city,
            ROUND(AVG(temperature_celsius), 2) AS avg_temp,
            ROUND(MIN(temperature_celsius), 2) AS min_temp,
            ROUND(MAX(temperature_celsius), 2) AS max_temp,
            COUNT(*) AS readings
        FROM weather_data
        GROUP BY city
        ORDER BY avg_temp DESC
    """).fetchdf()
    print(result)

    # Query 3: Hottest and coldest cities
    print("\n" + "=" * 50)
    print("Query 3: Current hottest and coldest cities")
    print("=" * 50)
    result = con.execute("""
        SELECT city, temperature_celsius, humidity, weather_description
        FROM weather_data
        WHERE data_timestamp = (
            SELECT MAX(data_timestamp) FROM weather_data
        )
        ORDER BY temperature_celsius DESC
    """).fetchdf()
    print(result)

What do these queries do?

Query 2 uses GROUP BY city to calculate the average, minimum, and maximum temperature for each city across all readings. COUNT(*) tells you how many readings each city has.
Query 3 uses a subquery to find the most recent timestamp, then filters to show only the latest reading for each city. This gives you a snapshot of current weather ranked from hottest to coldest.

Save queries.py and run it again:

python queries.py

You should now see three query results. Query 2 shows a table with each city's average, min, and max temperatures. Query 3 shows the cities ranked by their most recent temperature.

Seeing duplicate query output?

If Query 2 or Query 3 appears twice, you may have accidentally pasted the code below con.close() instead of above it. Make sure all queries sit between con = duckdb.connect(DB_PATH) and con.close().
Help me fix my queries.py file

Write a window function query

The queries so far have looked at each reading independently. A window function lets you compare a row to other rows in the same group without collapsing everything into a single aggregate.

The LAG function peeks at the previous row's value. You will use it to calculate how much the temperature changed between consecutive readings for each city.

In queries.py, add the following query above con.close():

    # Query 4: Temperature change using LAG window function
    print("\n" + "=" * 50)
    print("Query 4: Temperature change between readings (LAG)")
    print("=" * 50)
    result = con.execute("""
        SELECT
            city,
            data_timestamp,
            temperature_celsius,
            LAG(temperature_celsius) OVER (
                PARTITION BY city ORDER BY data_timestamp
            ) AS previous_temp,
            ROUND(
                temperature_celsius - LAG(temperature_celsius) OVER (
                    PARTITION BY city ORDER BY data_timestamp
                ), 2
            ) AS temp_change
        FROM weather_data
        ORDER BY city, data_timestamp
    """).fetchdf()
    print(result)

What does the LAG window function do?

PARTITION BY city tells DuckDB to treat each city as a separate group. ORDER BY data_timestamp sorts the readings within each group chronologically.

LAG(temperature_celsius) then grabs the temperature from the previous row in that group. The first reading for each city shows NULL because there is no earlier reading to compare against.

Why learn window functions?

Window functions are one of the most commonly tested SQL skills in data engineering interviews. They let you do running totals, rankings, and row-to-row comparisons that are impossible with plain GROUP BY.

Save queries.py and run it again:

python queries.py

You should see Query 4 output with columns for previous_temp and temp_change. The first reading for each city shows None for those columns.

Query 4 showing all NULLs?

If every row shows None for previous_temp, your database only has one batch of data per city. Run pipeline.py once more to add a second batch, then rerun queries.py.
Help me understand my LAG query results

Export a query result to CSV

The final piece of the data lifecycle is getting results out of the database and into a format other people can use. DuckDB has a built-in COPY command that exports query results directly to a CSV file.

In queries.py, add the following query above con.close():

    # Query 5: Export to CSV
    print("\n" + "=" * 50)
    print("Query 5: Exporting average temps to CSV...")
    print("=" * 50)
    con.execute("""
        COPY (
            SELECT
                city,
                ROUND(AVG(temperature_celsius), 2) AS avg_temp,
                ROUND(AVG(humidity), 2) AS avg_humidity,
                COUNT(*) AS total_readings
            FROM weather_data
            GROUP BY city
            ORDER BY avg_temp DESC
        ) TO 'weather_summary.csv' (HEADER, DELIMITER ',')
    """)
    print("  Exported to weather_summary.csv")

What does this query do?

The COPY command wraps any SELECT query and writes the results straight to a file. HEADER adds column names as the first row.

This query aggregates your weather data into a per-city summary with average temperature, average humidity, and total readings. The output lands in weather_summary.csv inside your project folder.

Save queries.py and run it one final time:

python queries.py

You should see all five query results print in sequence. The last line reads Exported to weather_summary.csv.
Confirm the CSV file was created by running:

ls

You should see weather_summary.csv listed alongside your other project files.

Don't see weather_summary.csv?

Make sure your terminal is inside the weather-etl folder. The CSV file is created in whatever directory Python runs from.
Check the terminal output for errors above the Query 5 section. A syntax error in the COPY statement can silently prevent the file from being created.
Help me debug my CSV export

✔️ Awesome, I've got everything!

Double check you have saved queries.py.

ⓧ I'd like to double check the full code

Here is the complete queries.py file. Compare it against yours and make sure everything matches.

import duckdb
from config import DB_PATH


def run_queries():
    """Run analytical SQL queries on the weather data."""
    con = duckdb.connect(DB_PATH)

    # Query 1: All data for a specific city
    print("=" * 50)
    print("Query 1: All weather data for London")
    print("=" * 50)
    result = con.execute("""
        SELECT city, temperature_celsius, humidity, weather_description, data_timestamp
        FROM weather_data
        WHERE city = 'London'
        ORDER BY data_timestamp DESC
    """).fetchdf()
    print(result)

    # Query 2: Average temperature per city
    print("\n" + "=" * 50)
    print("Query 2: Average temperature per city")
    print("=" * 50)
    result = con.execute("""
        SELECT
            city,
            ROUND(AVG(temperature_celsius), 2) AS avg_temp,
            ROUND(MIN(temperature_celsius), 2) AS min_temp,
            ROUND(MAX(temperature_celsius), 2) AS max_temp,
            COUNT(*) AS readings
        FROM weather_data
        GROUP BY city
        ORDER BY avg_temp DESC
    """).fetchdf()
    print(result)

    # Query 3: Hottest and coldest cities
    print("\n" + "=" * 50)
    print("Query 3: Current hottest and coldest cities")
    print("=" * 50)
    result = con.execute("""
        SELECT city, temperature_celsius, humidity, weather_description
        FROM weather_data
        WHERE data_timestamp = (
            SELECT MAX(data_timestamp) FROM weather_data
        )
        ORDER BY temperature_celsius DESC
    """).fetchdf()
    print(result)

    # Query 4: Temperature change using LAG window function
    print("\n" + "=" * 50)
    print("Query 4: Temperature change between readings (LAG)")
    print("=" * 50)
    result = con.execute("""
        SELECT
            city,
            data_timestamp,
            temperature_celsius,
            LAG(temperature_celsius) OVER (
                PARTITION BY city ORDER BY data_timestamp
            ) AS previous_temp,
            ROUND(
                temperature_celsius - LAG(temperature_celsius) OVER (
                    PARTITION BY city ORDER BY data_timestamp
                ), 2
            ) AS temp_change
        FROM weather_data
        ORDER BY city, data_timestamp
    """).fetchdf()
    print(result)

    # Query 5: Export to CSV
    print("\n" + "=" * 50)
    print("Query 5: Exporting average temps to CSV...")
    print("=" * 50)
    con.execute("""
        COPY (
            SELECT
                city,
                ROUND(AVG(temperature_celsius), 2) AS avg_temp,
                ROUND(AVG(humidity), 2) AS avg_humidity,
                COUNT(*) AS total_readings
            FROM weather_data
            GROUP BY city
            ORDER BY avg_temp DESC
        ) TO 'weather_summary.csv' (HEADER, DELIMITER ',')
    """)
    print("  Exported to weather_summary.csv")

    con.close()


if __name__ == "__main__":
    run_queries()

Secret mission

Add Data Quality Checks

A real production pipeline does not just move data. It validates it. Hiring managers specifically look for data quality awareness in portfolio projects. Can you add a validation layer that catches bad data before it enters your database?

Before you go

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. The OpenWeatherMap free tier does not require a credit card, and DuckDB stores data in a local file. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

OpenWeatherMap free account and API key.
Python packages: requests, pandas, duckdb.
Project folder with Python files: config.py, extract.py, transform.py, load.py, pipeline.py, queries.py.
DuckDB database file: weather.duckdb.
Exported CSV file: weather_summary.csv.

Keep everything running

No action needed. Choose this if you plan to keep running the pipeline to accumulate more weather data or want to experiment with new queries.

Your project folder, database file, and installed packages stay where they are.
Your OpenWeatherMap API key remains active on the free tier at no cost.

Pause - I'll come back to this later

Free up a small amount of disk space while keeping your project files intact for later.

Delete the weather.duckdb file to reclaim disk space. You can regenerate it by running pipeline.py again later.
Delete the weather_summary.csv file. You can regenerate it by running queries.py again.
Keep all .py files so your code is ready when you come back.
Your OpenWeatherMap account and API key stay active at no cost.

Delete - I don't want to use this again

Remove all project resources and start fresh.

Delete the entire project folder from your Desktop.
Uninstall the Python packages by running:

pip uninstall requests pandas duckdb -y

Optionally, delete your OpenWeatherMap account by visiting your account settings at openweathermap.org.

Mission accomplished

That's a Wrap!

Nice work! You just built a complete ETL pipeline from scratch. That is a real data engineering skill that teams use every single day to move data from sources into analytics-ready storage.

Here's what you learned:

Extracted live weather data from the OpenWeatherMap API and transformed messy JSON into a clean, structured pandas DataFrame with proper data types and unit conversions.
Loaded data into DuckDB with deduplication checks, and wrote analytical SQL queries including GROUP BY, ORDER BY, and the LAG window function.
Orchestrated the full Extract, Transform, Load flow with a single pipeline script and exported query results to CSV for reporting.
Secret Mission: Added data quality validation checks to catch bad data before it enters the database.

Ready to quiz yourself?