Build a Weather Data Pipeline

Build an ETL pipeline that loads weather API data into DuckDB.

Introduction

30 Second Summary

Every weather app, stock ticker, and recommendation engine depends on the same behind-the-scenes pattern: pulling raw data from a source, cleaning it up, and storing it somewhere useful. This pattern is called ETL (Extract, Transform, Load), and it is the single most important skill in data engineering.

In this project, you will build a complete ETL pipeline that pulls live weather forecast data from an API, transforms it with Python, and loads it into a database you can query with SQL.

What You'll Build

A Python data pipeline that extracts 7 days of hourly weather data from the Open-Meteo API for multiple cities, transforms it into clean pandas DataFrames, and loads the results into a local DuckDB analytical database.

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

  • A working Python ETL script (pipeline.py) that extracts live weather data from an API, transforms it, and loads it into a database.
  • A local DuckDB database file (weather.duckdb) containing structured weather data you can query with SQL.
  • Analytical SQL queries that answer real questions about temperature trends, precipitation patterns, and city comparisons.
  • Secret Mission: Add data quality checks that validate your pipeline output and flag anomalies before loading.

Are there any prerequisites?

This project assumes you have basic Python syntax knowledge (variables, functions, loops) and basic SQL experience (SELECT, WHERE, GROUP BY). You do not need prior experience with pandas, DuckDB, or APIs.

Set Up Your Environment

Every data pipeline needs a clean workspace with the right tools installed. Your pipeline will use three Python libraries to extract, transform, and load weather data.

In this step, you will create a project folder, set up an isolated Python environment, and install the packages you need.

In this step, get ready to:
  • Create your project directory and virtual environment.
  • Install DuckDB, pandas, and requests.
  • Verify all packages are working.
Create your project directory and virtual environment

A virtual environment keeps your project's packages separate from your system Python. This prevents conflicts between projects that need different library versions.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir weather-pipeline
cd weather-pipeline
  • Confirm the folder exists by running:
ls

You should see an empty directory listing (no files yet).

Now create a virtual environment inside your project folder.

macOS

  • Create and activate your virtual environment by running these commands:
python3 -m venv venv
source venv/bin/activate

Windows

  • Create and activate your virtual environment by running these commands:
python -m venv venv
venv\Scripts\activate

How do I know it worked?

After activation, your terminal prompt changes to show (venv) at the beginning of the line. This confirms you are now working inside the virtual environment.

Don't see (venv) in your prompt?

On Windows PowerShell, you may see an error about execution policies. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned and then try the activate command again.

On macOS, make sure you used source venv/bin/activate (not just venv/bin/activate without source).

Help me activate my virtual environment.

Install the project packages

Your pipeline needs three libraries. DuckDB is the analytical database where your data will live. pandas handles data transformation. The requests library handles HTTP calls to the weather API.

  • Install all three packages by running this command:
pip install duckdb==1.5.3 pandas==3.0.3 requests==2.34.2

Why pin specific versions?

The == syntax locks each package to an exact version. This ensures everyone following this project gets the same behaviour, regardless of when they run the install.

The install takes about 30 seconds. You should see progress bars for each package followed by a success message.

Seeing an error during installation?

If you see a message about incompatible Python version, this project requires Python 3.11 or higher. Check your version by running python --version (inside your venv) or python3 --version outside of it.

If your version is below 3.11, install a newer Python from python.org/downloads, then recreate your virtual environment using the new Python.

Help me fix my pip install error.

Verify the installation
  • Confirm all three packages are importable by running this command:
python -c "import duckdb; import pandas; import requests; print('All packages installed successfully')"

✔️ I see 'All packages installed successfully'

Your environment is ready. All three packages are installed and importable.

ⓧ I see a ModuleNotFoundError

This means one of the packages did not install correctly.

  • Check that your virtual environment is active (you should see (venv) in your prompt).
  • Run the pip install command again to retry the installation.
  • If the error persists, try installing the failing package individually, for example: pip install duckdb==1.5.3.

Help me debug my ModuleNotFoundError.

Your environment is ready to go. Next up, you will write the Extract phase of your pipeline and pull live weather data from an API.

Extract Weather Data from the API

Your environment is set up and ready to go. Now it's time to write the first phase of your ETL pipeline: the Extract.

The Extract phase pulls raw data from a source into your pipeline. In this project, your source is the Open-Meteo weather API, a free service that returns hourly forecast data as JSON.

In this step, get ready to:
  • Write a function that extracts weather data from the Open-Meteo API.
  • Pull forecast data for three cities (London, New York, Tokyo).
  • Handle network errors gracefully so the pipeline does not crash.
Create pipeline.py and add your imports

Every Python project starts with a file and the libraries it depends on. You will import four libraries that power different parts of the pipeline.

  • In your code editor, create a new file called pipeline.py inside your project folder.
  • Add the following imports at the top of the file:
import requests
import pandas as pd
import duckdb
from datetime import datetime

What do these imports do?

  • requests sends HTTP requests to the Open-Meteo API to fetch weather data.
  • pandas (imported as pd) transforms raw data into structured tables called DataFrames.
  • duckdb is the analytical database where you will load your transformed data.
  • datetime provides date and time utilities for working with timestamps.
  • Save pipeline.py.
  • Confirm the file exists by running:
ls

You should see pipeline.py listed in the output.

Don't see pipeline.py?

  • Make sure you saved the file after adding the imports.
  • Check that your terminal is in the same directory as the file. Run pwd to see your current location.

Help me find my pipeline.py file

Write the extract_weather() function

This function is the heart of the Extract phase. It loops through a dictionary of cities, calls the Open-Meteo forecast API for each one, and collects the responses into a list.

Why Open-Meteo?

Open-Meteo is completely free for non-commercial use with no API key or signup required. It returns rich hourly time-series data in clean JSON, making it perfect for learning ETL without authentication friction.

  • In pipeline.py, add the following function below your imports (leave one blank line between the imports and the function):
def extract_weather(cities: dict) -> list[dict]:
    """Extract hourly weather forecast data from Open-Meteo API for multiple cities."""
    all_data = []

    for city_name, coords in cities.items():
        url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            all_data.append({"city": city_name, "data": data})
            print(f"Extracted {len(data['hourly']['time'])} hourly records for {city_name}")
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data for {city_name}: {e}")

    return all_data

What does this function do?

  • cities.items() loops through a dictionary where each key is a city name and each value holds latitude/longitude coordinates.
  • The params dictionary tells the API which weather variables to return (temperature, humidity, precipitation, wind speed) and how many days of forecast to include.
  • requests.get() sends the HTTP request with a 10-second timeout so the script does not hang indefinitely.
  • response.raise_for_status() raises an error if the API returns an HTTP error code like 400.
  • The try/except block catches any network or API error. If one city fails, the function continues fetching the others instead of crashing.
  • Save pipeline.py.
  • Test the function by running this command in your terminal:
python -c "from pipeline import extract_weather; extract_weather({'London': {'lat': 51.51, 'lon': -0.13}})"

You should see output like: Extracted 168 hourly records for London.

Why 168 records?

The API returns one record per hour for 7 days. That is 24 hours multiplied by 7 days, which equals 168 hourly data points per city.

Seeing an error instead of records?

  • If you see ModuleNotFoundError: No module named 'requests', make sure your virtual environment is activated before running the command.
  • If you see a connection error, check your internet connection. The API requires network access.
  • If you see ModuleNotFoundError: No module named 'pipeline', make sure your terminal is in the same folder as pipeline.py.

Help me fix my extract_weather function

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is what your complete pipeline.py should look like at this point:

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


def extract_weather(cities: dict) -> list[dict]:
    """Extract hourly weather forecast data from Open-Meteo API for multiple cities."""
    all_data = []

    for city_name, coords in cities.items():
        url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            all_data.append({"city": city_name, "data": data})
            print(f"Extracted {len(data['hourly']['time'])} hourly records for {city_name}")
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data for {city_name}: {e}")

    return all_data

Your Extract phase is working. Next up, you will transform this raw JSON into a clean pandas DataFrame with proper column names and derived metrics.

Transform Data with pandas

Your Extract phase pulls raw JSON from the Open-Meteo API. The data arrives as nested arrays of floats and ISO timestamps, which is fine for an API response but terrible for analysis.

The Transform phase reshapes this raw data into a clean, analysis-ready table. You will use pandas to map API fields to readable column names, add columns that do not exist in the source data, and merge all cities into a single DataFrame.

In this step, get ready to:
  • Convert raw API JSON into a pandas DataFrame with proper column names.
  • Add city and date columns to enrich the data.
  • Calculate derived metrics like a feels-like temperature and a rain indicator.
Convert the raw API JSON into a DataFrame

The Open-Meteo API returns hourly data as separate arrays (one for temperatures, one for timestamps, one for humidity, etc.). A pandas DataFrame organises these into a single table where each row is one hour of weather data.

  • In pipeline.py, scroll to the bottom of the file (below the extract_weather() function).
  • Add the start of your transform function by copying this code:
def transform_weather(raw_data: list[dict]) -> pd.DataFrame:
    """Transform raw API responses into a clean pandas DataFrame."""
    frames = []

    for entry in raw_data:
        city_name = entry["city"]
        hourly = entry["data"]["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"],
            }
        )

What does this code do?

  • frames = [] creates an empty list that will hold one DataFrame per city.
  • The loop iterates through each city's raw data (the list of dicts your extract_weather() function returned).
  • pd.DataFrame() maps each API array to a readable column name. For example, temperature_2m becomes temperature_c.
  • pd.to_datetime() converts the raw ISO timestamp strings into proper datetime objects that pandas can sort and filter.
Add city and date columns

The API response does not include which city the data belongs to. You need to tag each row with its city name. You also need a date-only column for daily aggregations later.

  • Directly below the closing parenthesis of the pd.DataFrame() call, add these two lines (keep the same indentation level as the df = pd.DataFrame(...) line):
        df["city"] = city_name
        df["date"] = df["timestamp"].dt.date

What does this code do?

  • df["city"] = city_name adds a new column that stamps every row with the city it belongs to. This is essential for filtering and grouping later.
  • df["timestamp"].dt.date extracts just the date portion from each timestamp. You will use this column for daily aggregation queries in Step 5.
Calculate derived metrics and assemble the final DataFrame

Derived columns add analytical value that does not exist in the source data. A "feels like" temperature approximation and a rain indicator make downstream queries simpler.

  • Directly below the df["date"] line you just added, paste the following code to finish the function:
        df["feels_like_c"] = df["temperature_c"] - (df["wind_speed_kmh"] * 0.1)
        df["is_rainy"] = df["precipitation_mm"] > 0.1

        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)
    print(f"Transformed {len(combined)} total rows")
    return combined

What does this code do?

  • feels_like_c subtracts a wind chill factor (10% of wind speed) from the actual temperature. This is a simplified approximation, not a meteorological formula.
  • is_rainy is a boolean column. Any hour with more than 0.1mm of precipitation is flagged as rainy.
  • frames.append(df) adds the finished city DataFrame to the list. After the loop, pd.concat() merges all city DataFrames into one table.
  • ignore_index=True resets the row numbers so they run 0, 1, 2... across the combined table instead of restarting at 0 for each city.
  • Save pipeline.py.
  • You will not see output yet. The transform_weather() function is ready, but the next step (Load) will wire everything together and run the full pipeline.

Seeing a red underline or indentation error?

  • Make sure the df["city"], df["date"], df["feels_like_c"], and df["is_rainy"] lines are indented 8 spaces (two levels inside the function and for loop).
  • The combined = pd.concat(...) line must be at 4 spaces of indentation (inside the function but outside the for loop).
  • Check that you have a blank line between extract_weather() and def transform_weather(...) to separate the two functions.

Help me fix my indentation in the transform_weather function

✔️ 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 full pipeline.py file so far. Compare it to what you have:

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


def extract_weather(cities: dict) -> list[dict]:
    """Extract hourly weather forecast data from Open-Meteo API for multiple cities."""
    all_data = []

    for city_name, coords in cities.items():
        url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            all_data.append({"city": city_name, "data": data})
            print(f"Extracted {len(data['hourly']['time'])} hourly records for {city_name}")
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data for {city_name}: {e}")

    return all_data


def transform_weather(raw_data: list[dict]) -> pd.DataFrame:
    """Transform raw API responses into a clean pandas DataFrame."""
    frames = []

    for entry in raw_data:
        city_name = entry["city"]
        hourly = entry["data"]["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"] = city_name
        df["date"] = df["timestamp"].dt.date
        df["feels_like_c"] = df["temperature_c"] - (df["wind_speed_kmh"] * 0.1)
        df["is_rainy"] = df["precipitation_mm"] > 0.1

        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)
    print(f"Transformed {len(combined)} total rows")
    return combined

Your Transform phase is ready. It takes messy API JSON and outputs a clean, enriched DataFrame. Next up, you will load this DataFrame into a DuckDB database and run the full pipeline end-to-end.

Load Data into DuckDB

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

The problem is that DataFrames live in memory. When your script ends, the data disappears. In this step, you will load the transformed data into DuckDB, a columnar analytical database that persists your data to a file on disk.

In this step, get ready to:
  • Write a load function that creates a DuckDB table from your DataFrame.
  • Wire up the full ETL pipeline and run it end-to-end.
Write the load function

DuckDB can query pandas DataFrames directly by referencing the Python variable name in SQL statements. This means you can write SELECT * FROM df and DuckDB reads straight from your DataFrame without any manual row-by-row insertion.

  • In pipeline.py, add the following load_to_duckdb() function below your transform_weather() function:
def load_to_duckdb(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
    """Load the transformed DataFrame into a DuckDB database."""
    con = duckdb.connect(db_path)

    con.execute("DROP TABLE IF EXISTS hourly_weather")
    con.execute(
        """
        CREATE TABLE hourly_weather AS
        SELECT * FROM df
        """
    )

    row_count = con.execute("SELECT COUNT(*) FROM hourly_weather").fetchone()[0]
    print(f"Loaded {row_count} rows into hourly_weather table")

    con.close()

What does this code do?

  • duckdb.connect(db_path) opens (or creates) a database file at the given path. This is where your data persists on disk.
  • DROP TABLE IF EXISTS ensures a clean slate each time you run the pipeline. This makes the pipeline idempotent.
  • CREATE TABLE hourly_weather AS SELECT * FROM df creates the table directly from your pandas DataFrame. DuckDB reads df as a reference to the Python variable in scope.
  • The final SELECT COUNT(*) verifies the load succeeded by printing how many rows ended up in the table.
  • Save pipeline.py.

Why DROP TABLE before CREATE?

Without the DROP, running the pipeline a second time would fail because the table already exists. Dropping first means you can re-run the pipeline as many times as you want and always get a fresh, up-to-date table.

Wire up the full pipeline and run it

You now have all three phases of the ETL pipeline as separate functions. A main() function will tie them together in sequence: Extract, Transform, Load.

  • At the bottom of pipeline.py, below your load_to_duckdb() function, add the following main() function and entry point:
def main():
    cities = {
        "London": {"lat": 51.51, "lon": -0.13},
        "New York": {"lat": 40.71, "lon": -74.01},
        "Tokyo": {"lat": 35.68, "lon": 139.69},
    }

    print("=== EXTRACT ===")
    raw_data = extract_weather(cities)

    print("\n=== TRANSFORM ===")
    df = transform_weather(raw_data)

    print("\n=== LOAD ===")
    load_to_duckdb(df)


if __name__ == "__main__":
    main()

What does this code do?

  • The cities dictionary defines three locations with their latitude and longitude coordinates for the Open-Meteo API.
  • Each phase is called in order: extract_weather() fetches the raw data, transform_weather() cleans it, and load_to_duckdb() persists it.
  • The if __name__ == "__main__" guard ensures main() only runs when you execute the script directly, not when importing it as a module.
  • Save pipeline.py.
  • Run the full pipeline by running this command in your terminal:
python pipeline.py

You should see output showing all three phases completing in sequence, ending with a message like Loaded 504 rows into hourly_weather table.

Why 504 rows?

The Open-Meteo API returns 168 hourly records per city (24 hours x 7 days). With 3 cities, that gives you 3 x 168 = 504 rows. If you see this number, your pipeline extracted, transformed, and loaded everything correctly.

Seeing fewer than 504 rows?

  • If you see 0 rows or a connection error, check that your virtual environment is active and duckdb is installed.
  • If you see an error mentioning the DataFrame variable, make sure the parameter in load_to_duckdb() is named df to match the SQL query SELECT * FROM df.
  • If you see a network timeout error, check your internet connection. The Extract phase needs to reach the Open-Meteo API.

Help me debug my DuckDB pipeline output

  • Confirm the database file was created by running:
ls

You should see weather.duckdb listed alongside pipeline.py in the output.

✔️ 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

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


def extract_weather(cities: dict) -> list[dict]:
    """Extract hourly weather forecast data from Open-Meteo API for multiple cities."""
    all_data = []

    for city_name, coords in cities.items():
        url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            all_data.append({"city": city_name, "data": data})
            print(f"Extracted {len(data['hourly']['time'])} hourly records for {city_name}")
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data for {city_name}: {e}")

    return all_data


def transform_weather(raw_data: list[dict]) -> pd.DataFrame:
    """Transform raw API responses into a clean pandas DataFrame."""
    frames = []

    for entry in raw_data:
        city_name = entry["city"]
        hourly = entry["data"]["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"] = city_name
        df["date"] = df["timestamp"].dt.date
        df["feels_like_c"] = df["temperature_c"] - (df["wind_speed_kmh"] * 0.1)
        df["is_rainy"] = df["precipitation_mm"] > 0.1

        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)
    print(f"Transformed {len(combined)} total rows")
    return combined


def load_to_duckdb(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
    """Load the transformed DataFrame into a DuckDB database."""
    con = duckdb.connect(db_path)

    con.execute("DROP TABLE IF EXISTS hourly_weather")
    con.execute(
        """
        CREATE TABLE hourly_weather AS
        SELECT * FROM df
        """
    )

    row_count = con.execute("SELECT COUNT(*) FROM hourly_weather").fetchone()[0]
    print(f"Loaded {row_count} rows into hourly_weather table")

    con.close()


def main():
    cities = {
        "London": {"lat": 51.51, "lon": -0.13},
        "New York": {"lat": 40.71, "lon": -74.01},
        "Tokyo": {"lat": 35.68, "lon": 139.69},
    }

    print("=== EXTRACT ===")
    raw_data = extract_weather(cities)

    print("\n=== TRANSFORM ===")
    df = transform_weather(raw_data)

    print("\n=== LOAD ===")
    load_to_duckdb(df)


if __name__ == "__main__":
    main()

Your data is now persisted in a real database file. Next up, you will write SQL queries against weather.duckdb to find temperature trends, rainy days, and city comparisons.

Query Your Data with SQL

Your data is sitting in DuckDB and ready to answer questions. The real value of a data pipeline is what happens after the Load phase.

In this step, you will write SQL queries that extract insights from your weather data. You will use window functions, aggregations, and DuckDB's native PIVOT syntax to compare cities side-by-side.

In this step, get ready to:
  • Write a window function query to find peak temperatures per city.
  • Calculate daily aggregates with GROUP BY.
  • Build a pivot comparison across all three cities.
Find the hottest forecast hour for each city

The first query answers: which single hour in the forecast has the highest temperature for each city? A regular GROUP BY cannot do this because you need both the row with the maximum temperature AND its timestamp.

This is where window functions come in. They let you rank rows within groups without collapsing the group into a single row.

  • In pipeline.py, find the main() function.
  • Add the following new function directly above main():
def query_results(db_path: str = "weather.duckdb") -> None:
    """Run analytical queries against the loaded data."""
    con = duckdb.connect(db_path)

    print("\n--- Hottest Forecast Hour Per City ---")
    result = con.execute(
        """
        WITH ranked AS (
            SELECT
                city,
                timestamp,
                temperature_c,
                ROW_NUMBER() OVER (PARTITION BY city ORDER BY temperature_c DESC) AS rn
            FROM hourly_weather
        )
        SELECT city, timestamp, temperature_c
        FROM ranked
        WHERE rn = 1
        """
    ).fetchdf()
    print(result.to_string(index=False))

What does this query do?

The WITH ranked AS (...) block creates a Common Table Expression (CTE) that numbers each row within its city group.

ROW_NUMBER() OVER (PARTITION BY city ORDER BY temperature_c DESC) assigns rank 1 to the hottest hour in each city. The outer query then filters to only those rank-1 rows, giving you the single hottest forecast hour per city.

You won't see any change yet. The next two queries will complete the function before you run it.

Calculate daily average temperature and total precipitation

The second query summarises hourly data into daily totals. This is the bread-and-butter of analytical SQL: grouping rows by a dimension (city and date) and computing aggregates.

  • In pipeline.py, add the following code directly below the last print(result.to_string(index=False)) line you just wrote, still inside the query_results() function:
    print("\n--- Daily Averages Per City ---")
    result = con.execute(
        """
        SELECT
            city,
            date,
            ROUND(AVG(temperature_c), 1) AS avg_temp_c,
            ROUND(SUM(precipitation_mm), 1) AS total_precip_mm
        FROM hourly_weather
        GROUP BY city, date
        ORDER BY city, date
        """
    ).fetchdf()
    print(result.to_string(index=False))

What does this query do?

GROUP BY city, date collapses 24 hourly rows into one row per city per day.

AVG(temperature_c) computes the mean temperature across those 24 hours. SUM(precipitation_mm) gives total rainfall for the day. ROUND(..., 1) keeps the output clean with one decimal place.

You won't see output yet. One more query to go.

Compare cities side-by-side and run the full pipeline

The third query reshapes the data so each city becomes its own column. DuckDB supports native PIVOT syntax that most databases do not offer, making this kind of comparison trivial.

  • Still inside query_results(), add the following code below the daily averages query:
    print("\n--- City Temperature Comparison (Daily Avg) ---")
    result = con.execute(
        """
        PIVOT (
            SELECT
                date,
                city,
                ROUND(AVG(temperature_c), 1) AS avg_temp
            FROM hourly_weather
            GROUP BY date, city
        )
        ON city
        USING FIRST(avg_temp)
        ORDER BY date
        """
    ).fetchdf()
    print(result.to_string(index=False))

    con.close()

What does this query do?

The inner SELECT calculates one average temperature per city per day. PIVOT ... ON city then takes the distinct city values and turns each one into its own column.

USING FIRST(avg_temp) tells DuckDB which value to place in each pivoted cell. The result is one row per date with London, New York, and Tokyo temperatures side-by-side.

Now connect query_results() to your pipeline by updating main().

  • In pipeline.py, scroll down to main() and find this line:
    load_to_duckdb(df)
  • Add the following two lines directly below it:
    print("\n=== QUERY ===")
    query_results()

What does this addition do?

This adds a fourth phase (QUERY) to your pipeline. After extracting, transforming, and loading, the pipeline now automatically runs your three analytical queries and prints the results.

  • Save pipeline.py.
  • Run the full pipeline by running this command in your terminal:
python pipeline.py

You should see all four phases printed: EXTRACT, TRANSFORM, LOAD, then QUERY with three formatted tables showing the hottest hours, daily averages, and city comparison.

Not seeing query results?

  • Make sure con.close() is at the very end of query_results(), after all three print statements. Closing the connection too early means later queries have no database to talk to.
  • Check that query_results() is defined as its own function at module level (no extra indentation), not nested inside load_to_duckdb().
  • If you see an error about the table not existing, the Load phase did not run first. Confirm your main() calls load_to_duckdb(df) before query_results().

Help me debug my query_results() function.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and all four phases print correctly.

ⓧ I'd like to double check the full code

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


def extract_weather(cities: dict) -> list[dict]:
    """Extract hourly weather forecast data from Open-Meteo API for multiple cities."""
    all_data = []

    for city_name, coords in cities.items():
        url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "forecast_days": 7,
            "timezone": "auto",
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            all_data.append({"city": city_name, "data": data})
            print(f"Extracted {len(data['hourly']['time'])} hourly records for {city_name}")
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data for {city_name}: {e}")

    return all_data


def transform_weather(raw_data: list[dict]) -> pd.DataFrame:
    """Transform raw API responses into a clean pandas DataFrame."""
    frames = []

    for entry in raw_data:
        city_name = entry["city"]
        hourly = entry["data"]["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"] = city_name
        df["date"] = df["timestamp"].dt.date
        df["feels_like_c"] = df["temperature_c"] - (df["wind_speed_kmh"] * 0.1)
        df["is_rainy"] = df["precipitation_mm"] > 0.1

        frames.append(df)

    combined = pd.concat(frames, ignore_index=True)
    print(f"Transformed {len(combined)} total rows")
    return combined


def load_to_duckdb(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
    """Load the transformed DataFrame into a DuckDB database."""
    con = duckdb.connect(db_path)

    con.execute("DROP TABLE IF EXISTS hourly_weather")
    con.execute(
        """
        CREATE TABLE hourly_weather AS
        SELECT * FROM df
        """
    )

    row_count = con.execute("SELECT COUNT(*) FROM hourly_weather").fetchone()[0]
    print(f"Loaded {row_count} rows into hourly_weather table")

    con.close()


def query_results(db_path: str = "weather.duckdb") -> None:
    """Run analytical queries against the loaded data."""
    con = duckdb.connect(db_path)

    print("\n--- Hottest Forecast Hour Per City ---")
    result = con.execute(
        """
        WITH ranked AS (
            SELECT
                city,
                timestamp,
                temperature_c,
                ROW_NUMBER() OVER (PARTITION BY city ORDER BY temperature_c DESC) AS rn
            FROM hourly_weather
        )
        SELECT city, timestamp, temperature_c
        FROM ranked
        WHERE rn = 1
        """
    ).fetchdf()
    print(result.to_string(index=False))

    print("\n--- Daily Averages Per City ---")
    result = con.execute(
        """
        SELECT
            city,
            date,
            ROUND(AVG(temperature_c), 1) AS avg_temp_c,
            ROUND(SUM(precipitation_mm), 1) AS total_precip_mm
        FROM hourly_weather
        GROUP BY city, date
        ORDER BY city, date
        """
    ).fetchdf()
    print(result.to_string(index=False))

    print("\n--- City Temperature Comparison (Daily Avg) ---")
    result = con.execute(
        """
        PIVOT (
            SELECT
                date,
                city,
                ROUND(AVG(temperature_c), 1) AS avg_temp
            FROM hourly_weather
            GROUP BY date, city
        )
        ON city
        USING FIRST(avg_temp)
        ORDER BY date
        """
    ).fetchdf()
    print(result.to_string(index=False))

    con.close()


def main():
    cities = {
        "London": {"lat": 51.51, "lon": -0.13},
        "New York": {"lat": 40.71, "lon": -74.01},
        "Tokyo": {"lat": 35.68, "lon": 139.69},
    }

    print("=== EXTRACT ===")
    raw_data = extract_weather(cities)

    print("\n=== TRANSFORM ===")
    df = transform_weather(raw_data)

    print("\n=== LOAD ===")
    load_to_duckdb(df)

    print("\n=== QUERY ===")
    query_results()


if __name__ == "__main__":
    main()

Secret mission

Add Data Quality Checks

Your pipeline works, but what happens when the API returns garbage data? In production, bad data silently corrupts your database. Add a validation gate that catches problems before they reach DuckDB.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project runs entirely locally, so there are no ongoing costs. Your files stay on your machine until you choose to remove them.

Resources you used:

  • Your project folder on the Desktop (contains pipeline.py, weather.duckdb, and your virtual environment with installed packages).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your pipeline, adding more cities, or extending your SQL queries.

  • Your project folder stays on your Desktop with all files intact.
  • You can re-run pipeline.py anytime to refresh the weather data.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files remain on disk and are ready to pick up again later.

  • If your virtual environment is still active, deactivate it by running:
deactivate
  • When you come back, navigate to your project folder and reactivate the virtual environment to resume.

Delete - I don't want to use this again

Remove all project files from your machine. This deletes your pipeline script, the DuckDB database, and the virtual environment with its installed packages.

  • If your virtual environment is still active, deactivate it first by running:
deactivate
  • Delete your project folder from the Desktop by running this command (macOS/Linux):
rm -rf ~/Desktop/[[PROJECT_FOLDER="your-project-folder-name"]]
  • On Windows, run this command instead:
rmdir /s /q %USERPROFILE%\Desktop\[[PROJECT_FOLDER="your-project-folder-name"]]
  • Confirm the folder is gone by checking your Desktop. You should no longer see the project folder listed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a complete ETL pipeline that pulls live weather data from an API, transforms it with pandas, and loads it into a columnar analytical database.

You've learned how to:

  • Built a Python ETL pipeline that extracts live weather forecast data from the Open-Meteo API for multiple cities.
  • Transformed raw JSON into clean pandas DataFrames and loaded them into DuckDB, a modern columnar analytical database.
  • Wrote analytical SQL queries using window functions, aggregations, and DuckDB's native PIVOT syntax to compare cities side-by-side.
  • Secret Mission: Added production-grade data quality checks that validate pipeline output and halt loading when anomalies are detected.

Ready to quiz yourself?