Weather Analytics Pipeline in Python

Build an ETL pipeline that loads weather data into a PostgreSQL star schema.

Introduction

30 Second Summary

Weather apps pull data from dozens of sources, transform it, and present clean forecasts on your phone. Behind the scenes, that's a data pipeline at work. The same pattern powers everything from financial dashboards to recommendation engines.

In this project, you will build a complete ETL pipeline that extracts live weather data from an API, loads it into PostgreSQL, and transforms it into a star schema designed for analytics.

What You'll Build

You run a single Python script that pulls real hourly weather data for 5 cities, loads it into your database, and transforms it into a clean star schema you can query with powerful analytical SQL.

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

  • A Python ETL pipeline that extracts live weather data from 5 cities and loads it into PostgreSQL with one command.
  • A star schema data warehouse with dimension tables for locations and dates, plus a fact table for weather observations that you can browse in pgAdmin.
  • A set of analytical SQL queries using window functions and CTEs that answer questions like "which city had the biggest temperature swing today?".
  • Secret Mission: Build a live Streamlit dashboard that visualizes your pipeline output with interactive charts.

Are there any prerequisites?

This project assumes you have basic Python knowledge and basic SQL knowledge. You also need PostgreSQL and pgAdmin already installed on your machine.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters. This is the moment you commit to the goal so every hands-on step that follows has clear purpose.

Set Up Your Pipeline Environment

Every data pipeline starts with a clean workspace. You need isolated Python packages so your pipeline dependencies don't clash with anything else on your machine. You also need a dedicated database to land your weather data in.

In this step, you will create a project folder, set up a Python virtual environment with the packages your pipeline needs, and create the weather_pipeline database in PostgreSQL.

In this step, get ready to:
  • Create a Python virtual environment and install pipeline packages.
  • Create the weather_pipeline database in pgAdmin.
  • Test the database connection with a Python script.
Create your project folder and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This means the libraries you install here stay separate from any other Python projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Open the integrated terminal by pressing Ctrl+` (macOS and Windows).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir weather-pipeline
cd weather-pipeline
  • Confirm the folder was created by running:
ls

The folder is empty for now. Your terminal prompt should show weather-pipeline as the current directory.

  • Now open this folder in VS Code so the file explorer shows your project. Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the weather-pipeline folder. Click Open.

Why reopen the folder?

Opening the folder tells VS Code to treat it as your workspace. The Explorer panel on the left now shows your project files, and any new terminal will start inside this folder automatically.

  • Open a fresh terminal by pressing Ctrl+` (macOS and Windows).
  • Create a virtual environment by running this command:
python -m venv venv

Seeing 'command not found'?

On macOS or Linux, the command may be python3 instead of python. Try running python3 -m venv venv instead.

Help me find the right Python command

  • Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Install the two Python packages your pipeline needs by running:
pip install requests psycopg2-binary

What do these packages do?

The requests library lets your Python script call web APIs and get data back. The psycopg2-binary library is a PostgreSQL adapter that lets Python connect to and query your database.

Pip will download and install both packages. You should see output ending with "Successfully installed" followed by the package names.

  • Verify both packages installed correctly by running:
python -c "import requests; import psycopg2; print('Ready')"

You should see Ready printed to your terminal. This confirms both libraries are importable.

Seeing an ImportError?

Make sure your virtual environment is active. You should see (venv) in your prompt. If not, re-run the activation command for your operating system.

If you activated the environment but still get an error, re-run pip install requests psycopg2-binary and check for error messages during installation.

Help me fix this ImportError

Create the pipeline database in pgAdmin

Your pipeline needs its own database to keep weather data separate from system databases. You will create a dedicated database called weather_pipeline using pgAdmin.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type pgAdmin and press Enter to open it.
  • In the left Browser panel, expand Servers to reveal your local PostgreSQL server.
  • If prompted for a password, enter the password you set when you installed PostgreSQL.
  • Right-click on Databases under your server.
  • Select Create, then Database.
  • In the Database field, type weather_pipeline.
  • Click Save.

You should see weather_pipeline appear under Databases in the left panel. This is where all your pipeline data will live.

Don't see the new database?

Right-click on Databases and select Refresh. The new database should appear in the list.

If you got a permission error when saving, make sure you are connected as the postgres user. Check the server properties to confirm.

Help me troubleshoot my database creation

Test the database connection

Now you will write a short Python script that proves your code can talk to the new database. If this works, your pipeline environment is fully ready.

  • Switch back to VS Code.
  • Create a new file by clicking File in the top menu bar, then New File.
  • Save it immediately with Cmd+S (macOS) or Ctrl+S (Windows). Name it test_connection.py.
  • Copy the following code into test_connection.py:
import psycopg2

# Connection details for your local PostgreSQL database
DB_CONFIG = {
    "dbname": "weather_pipeline",
    "user": "postgres",
    "password": "your-password-here",
    "host": "localhost",
    "port": "5432",
}

# Connect and print the server version
conn = psycopg2.connect(**DB_CONFIG)
cursor = conn.cursor()
cursor.execute("SELECT version();")
version = cursor.fetchone()[0]
print(f"Connected! PostgreSQL version: {version}")
cursor.close()
conn.close()

What does this code do?

The DB_CONFIG dictionary stores the connection details for your local PostgreSQL server. These match the defaults from a standard PostgreSQL installation.

The script connects using psycopg2.connect(), runs a query to fetch the server version, prints it, and then cleanly closes the connection.

  • Replace your-password-here with the password you use to log into your local PostgreSQL server (the same one pgAdmin asked for).
  • Save the file with Cmd+S (macOS) or Ctrl+S (Windows).

Before you run this, what do you think will happen? Will the script succeed, or will it fail?

  • Run the script in your terminal by running:
python test_connection.py

You should see output like:

Connected! PostgreSQL version: PostgreSQL 16.x ...

The exact version number depends on your installation. Seeing any version string means your Python environment can successfully connect to weather_pipeline.

Seeing a connection error?

If you see a "password authentication failed" error, double-check the password in DB_CONFIG matches what you entered in pgAdmin.

If you see "connection refused", make sure the PostgreSQL service is running. On Windows, search for Services and check that the PostgreSQL service is started. On macOS, check that Postgres.app is running or the Homebrew service is active.

Help me fix this connection error

✔️ Awesome, I've got everything!

Great. Double-check you have saved test_connection.py and that your terminal output shows a PostgreSQL version string.

ⓧ I'd like to double check the full code

Here is the complete test_connection.py file. Make sure yours matches (with your own password filled in):

import psycopg2

# Connection details for your local PostgreSQL database
DB_CONFIG = {
    "dbname": "weather_pipeline",
    "user": "postgres",
    "password": "your-password-here",
    "host": "localhost",
    "port": "5432",
}

# Connect and print the server version
conn = psycopg2.connect(**DB_CONFIG)
cursor = conn.cursor()
cursor.execute("SELECT version();")
version = cursor.fetchone()[0]
print(f"Connected! PostgreSQL version: {version}")
cursor.close()
conn.close()

Your pipeline environment is locked in. You have isolated packages, a dedicated database, and a proven connection between the two. Next up, you will pull live weather data from a real API into this environment.

Extract Weather Data from the Open-Meteo API

Your environment is connected and ready. Now it is time to build the first real layer of your pipeline: the extraction layer that pulls live data from the outside world.

Before you can transform or model data, you need a working extraction function that reaches out to a real API and brings structured data back. In this step, you will call the Open-Meteo Forecast API for 5 cities and parse the response into clean Python dictionaries.

In this step, get ready to:
  • Build an extraction function that calls the Open-Meteo API for 5 cities.
  • Add data validation and logging to confirm the extraction worked.
  • Run the script and see 840 rows of weather data printed to your terminal.
Build the extraction function

The Open-Meteo API is a free weather forecast service that requires no API key and no signup. You send it a latitude, longitude, and a list of weather variables, and it returns hourly forecast data as JSON.

What is data extraction in an ETL pipeline?

Extraction is the "E" in ETL (Extract, Transform, Load). It means pulling raw data from a source system into your pipeline.

Your source is a REST API. In production, sources could be databases, flat files, message queues, or third-party SaaS tools. The extraction layer handles all the connection logic so the rest of your pipeline only deals with clean data.

  • Create a new file called extract.py in your project folder by running this command in the VS Code terminal:
touch extract.py
  • Open extract.py in the VS Code editor by clicking on it in the Explorer sidebar.
  • Add the following code to define your city coordinates, the API endpoint, and the start of your extraction function:
import requests

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

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


def extract_weather_data():
    all_records = []

    for city_name, coords in CITIES.items():
        # Request hourly weather metrics for this city
        params = {
            "latitude": coords["lat"],
            "longitude": coords["lon"],
            "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
            "timezone": "auto",
        }
        response = requests.get(API_URL, params=params)
        response.raise_for_status()
        data = response.json()

        hourly = data["hourly"]
        times = hourly["time"]

What does this code do?

  • CITIES stores the latitude and longitude for each city. Open-Meteo uses coordinates instead of city names.
  • API_URL points to the Open-Meteo forecast endpoint. No API key is needed.
  • The params dictionary asks for four hourly variables: temperature at 2m, relative humidity at 2m, precipitation, and wind speed at 10m.
  • response.raise_for_status() throws an error if the API returns a non-200 status code, so you know immediately if something went wrong.

The function is not complete yet. You will add the record-building logic and validation next.

Add basic data validation

The API call brings back raw JSON, but you need to validate the response and reshape it into clean records your pipeline can use downstream.

  • In extract.py, add the following code directly below the times = hourly["time"] line, at the same indentation level (inside the for loop):
        # Skip cities with no data
        if not times:
            print(f"WARNING: No data returned for {city_name}")
            continue

        # Build a record for each hour of data
        for i in range(len(times)):
            record = {
                "city_name": city_name,
                "latitude": coords["lat"],
                "longitude": coords["lon"],
                "timestamp": times[i],
                "temperature_c": hourly["temperature_2m"][i],
                "relative_humidity": hourly["relative_humidity_2m"][i],
                "precipitation_mm": hourly["precipitation"][i],
                "wind_speed_kmh": hourly["wind_speed_10m"][i],
            }
            all_records.append(record)

        print(f"Extracted {len(times)} records for {city_name}")

    # Print summary after all cities are processed
    print(f"\nTotal records extracted: {len(all_records)}")
    print(f"Sample record: {all_records[0]}")
    return all_records


if __name__ == "__main__":
    extract_weather_data()

What does this code do?

  • if not times: checks that the API actually returned data for the city. If the array is empty, it logs a warning and skips to the next city.
  • The inner for loop builds one dictionary per hour, mapping API field names (like temperature_2m) to clean pipeline names (like temperature_c).
  • The print statements log how many records each city contributed. This is your extraction observability.
  • if __name__ == "__main__": lets you run this file directly as a script, but also import the function from other files later.
  • Save extract.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved extract.py before continuing.

ⓧ I'd like to double check the full code

Scroll up and confirm your file has these sections, in order:

  • The import statement and CITIES dictionary with 5 cities.
  • The API_URL constant pointing to https://api.open-meteo.com/v1/forecast.
  • The extract_weather_data() function with the API call, validation check, record-building loop, and print statements.
  • The if __name__ == "__main__": block at the bottom.
Run the extraction

Before you run this, think about what you expect to see. The API returns 7 days of hourly data by default. That is 168 hours. You are calling it for 5 cities. How many total records should that produce?

  • Run your extraction script by executing this command in the terminal:
python extract.py

You should see output like this (your actual temperatures will differ based on the current forecast):

Extracted 168 records for London
Extracted 168 records for New York
Extracted 168 records for Tokyo
Extracted 168 records for Sydney
Extracted 168 records for Berlin

Total records extracted: 840
Sample record: {'city_name': 'London', 'latitude': 51.5074, ...}

168 hours multiplied by 5 cities gives you exactly 840 records. Your extraction layer is working.

Not seeing 840 records?

  • If you see ModuleNotFoundError: No module named 'requests', make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
  • If you see a connection error, check that you have internet access and no firewall is blocking outbound HTTPS requests.
  • If you see an indentation error, make sure the if not times: block and the inner for loop are indented to the same level (two levels in, inside the outer for loop).
  • Help me debug my extract.py script

Why 840 records?

The Open-Meteo API returns 7 days of hourly data by default (168 hours). With 5 cities, that gives 168 x 5 = 840 hourly weather observations. This is a meaningful volume for learning dimensional modeling without being unwieldy.

Your extraction layer is pulling live weather data from 5 cities across the world. Next up, you will load this raw data into a PostgreSQL staging table so it is persisted and ready for transformation.

Load Raw Data into a Staging Table

Your extraction script can pull 840 rows of live weather data from the API. But right now, that data only exists in memory while the script runs.

In production pipelines, you never transform data in-flight. You land the raw data in a staging table first. This gives you a checkpoint you can inspect, debug, and re-process without hitting the API again. Land first, transform second.

In this step, get ready to:
  • Create a staging schema and table in your PostgreSQL database.
  • Load all 840 extracted rows into the staging table using Python.
  • Observe the flat table's limitations for analytics queries.
Create the staging schema and table

A staging table is a landing zone for raw data. It mirrors the shape of the source data exactly, with no transformations applied.

  • Create a new file called load.py in your project folder (the same folder where extract.py lives).
  • Add the imports, database configuration, and staging table creation function by pasting this code:
import psycopg2
from datetime import datetime, timezone
from extract import extract_weather_data


# Database connection settings
DB_CONFIG = {
    "dbname": "weather_pipeline",
    "user": "postgres",
    "password": "your-password-here",
    "host": "localhost",
    "port": "5432",
}


def create_staging_table(cursor):
    # Create a dedicated schema for raw/staging data
    cursor.execute("CREATE SCHEMA IF NOT EXISTS staging;")
    # Create the staging table with columns matching the API output
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS staging.raw_weather (
            id SERIAL PRIMARY KEY,
            city_name VARCHAR(100),
            latitude DECIMAL(8, 4),
            longitude DECIMAL(8, 4),
            timestamp VARCHAR(50),
            temperature_c DECIMAL(5, 2),
            relative_humidity INTEGER,
            precipitation_mm DECIMAL(6, 2),
            wind_speed_kmh DECIMAL(6, 2),
            loaded_at TIMESTAMP DEFAULT NOW()
        );
    """)

What does this code do?

This sets up the structure your raw data will land in.

  • from extract import extract_weather_data imports the extraction function you built in the previous step so this script can call it directly.
  • DB_CONFIG holds the connection details for your weather_pipeline database. Replace the password with the one you set for your PostgreSQL postgres user.
  • CREATE SCHEMA IF NOT EXISTS staging creates a separate namespace so your raw data lives apart from future analytics tables.
  • loaded_at TIMESTAMP DEFAULT NOW() automatically records when each row was inserted, so you know when data arrived.
  • Replace your-password-here in DB_CONFIG with your actual PostgreSQL password (the same one you used in test_connection.py).
  • Save load.py. You should see it appear in VS Code's explorer sidebar next to extract.py.

Don't see load.py in the sidebar?

  • Make sure you saved the file inside the same folder where extract.py lives. If it is in a different location, the from extract import extract_weather_data line will fail later.
  • Check that the filename is exactly load.py with no extra spaces or capital letters.

help me debug why my load.py file isn't showing up in VS Code

Load the extracted data

The load_data function will connect to the database, prepare the staging table, and bulk-insert every record the extraction function returns.

  • Add the first part of the load_data function below your create_staging_table function in load.py:
def load_data(records):
    # Connect to the database
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()

    # Ensure the staging table exists
    create_staging_table(cursor)

    # Clear any previous data so we get a fresh load
    cursor.execute("TRUNCATE TABLE staging.raw_weather RESTART IDENTITY;")

    # SQL template for inserting one row
    insert_sql = """
        INSERT INTO staging.raw_weather
            (city_name, latitude, longitude, timestamp,
             temperature_c, relative_humidity, precipitation_mm, wind_speed_kmh)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
    """

What's happening here?

  • TRUNCATE TABLE ... RESTART IDENTITY clears any previously loaded data so you always start fresh. This makes the pipeline idempotent (safe to re-run).
  • insert_sql defines the SQL template with %s placeholders. psycopg2 will safely substitute your Python values into these placeholders.
  • Complete the load_data function and add the script entry point by pasting this directly below the insert_sql definition:
    # Transform each dict into a tuple for the INSERT statement
    rows = [
        (
            r["city_name"],
            r["latitude"],
            r["longitude"],
            r["timestamp"],
            r["temperature_c"],
            r["relative_humidity"],
            r["precipitation_mm"],
            r["wind_speed_kmh"],
        )
        for r in records
    ]

    # Bulk-insert all rows in one call
    cursor.executemany(insert_sql, rows)
    conn.commit()

    # Verify the load by counting rows
    cursor.execute("SELECT COUNT(*) FROM staging.raw_weather;")
    count = cursor.fetchone()[0]
    print(f"\nLoaded {count} records into staging.raw_weather")

    cursor.close()
    conn.close()


if __name__ == "__main__":
    records = extract_weather_data()
    load_data(records)

What does this code do?

  • The list comprehension converts each dictionary from extract_weather_data() into a tuple that matches the column order of the INSERT statement.
  • cursor.executemany(insert_sql, rows) runs the INSERT statement once for every tuple in the list, loading all 840 records.
  • The if __name__ block calls extract_weather_data() first, then passes the results to load_data(). One script run does the full extract-then-load cycle.

Before you run this, how many records do you expect to see loaded into the staging table? Think back to what the extraction script printed.

  • Save load.py and run it in your terminal:
python load.py

You should see the familiar extraction output (168 records per city), followed by a new line confirming the load:

Extracted 168 records for London
Extracted 168 records for New York
Extracted 168 records for Tokyo
Extracted 168 records for Sydney
Extracted 168 records for Berlin

Total records extracted: 840
Sample record: {...}

Loaded 840 records into staging.raw_weather

Seeing a connection error or import error?

  • If you see ModuleNotFoundError: No module named 'extract', make sure load.py is in the same folder as extract.py and you are running the command from that folder.
  • If you see a password authentication error, double-check the password value in DB_CONFIG matches your PostgreSQL password.
  • If you see connection refused, confirm your PostgreSQL service is running. On Windows, check **Services** for the PostgreSQL entry.

help me fix my load.py connection or import error

✔️ Awesome, I've got everything!

Great. Double check you have saved load.py and that the output shows 840 loaded records.

ⓧ I'd like to double check the full code

Scroll up and confirm your load.py has these sections in order:

  • Imports (psycopg2, datetime, extract_weather_data).
  • DB_CONFIG dictionary with your database details.
  • create_staging_table(cursor) function that creates the schema and table.
  • load_data(records) function that truncates, inserts, and prints the count.
  • if __name__ == "__main__" block that calls extract_weather_data() then load_data().
Verify the staging table in pgAdmin

The data is now in PostgreSQL. Open pgAdmin and confirm it landed correctly.

  • In pgAdmin, expand your weather_pipeline database in the left sidebar.
  • Click the **Query Tool** button in the top toolbar (or right-click the database and select **Query Tool**).
  • Run this query to count your loaded rows:
SELECT COUNT(*) FROM staging.raw_weather;

You should see 840 in the results. That confirms every extracted row made it into the database.

  • Run this query to inspect a few sample rows:
SELECT * FROM staging.raw_weather LIMIT 5;

You should see rows with city names, coordinates, timestamps, and weather measurements.

Now try answering a simple analytics question: what is the average temperature per city? Run this:

SELECT city_name, AVG(temperature_c) AS avg_temp
FROM staging.raw_weather
GROUP BY city_name;

It works. You get results. But look at the underlying data more carefully.

  • Run this query to see the duplication problem:
SELECT city_name, latitude, longitude, COUNT(*)
FROM staging.raw_weather
GROUP BY city_name, latitude, longitude;

Why is this table structure a problem?

Each city's name, latitude, and longitude are repeated 168 times (once per hour). That is 840 rows of duplicated location data. If you needed to update London's coordinates, you would have to update 168 rows.

There is also no clean way to group by date or hour without string parsing the timestamp column. Asking "which city had the widest temperature swing today?" requires ugly string slicing and self-joins.

This flat structure works for simple lookups but becomes painful for real analytics. The next step fixes this with a star schema that separates facts from dimensions.

Your raw data is safely staged in PostgreSQL. Next up, you will transform this flat table into a proper star schema that separates locations, dates, and weather measurements into clean, linked tables.

Transform into a Star Schema

In the previous step, you loaded 840 rows of weather data into a flat staging table. You saw the problem firsthand: city names, latitude, and longitude repeat in every single row.

Answering a question like "which city had the biggest temperature swing today?" against this flat table requires messy string slicing and self-joins. A star schema solves this by separating facts from dimensions. One row per location, one row per date, and a lean fact table pointing to both. Queries become clean, fast, and readable.

In this step, get ready to:
  • Design and create a star schema with dimension and fact tables.
  • Populate dimension tables from your staging data.
  • Populate the fact table by joining staging to both dimensions.
Design and create the star schema

Your star schema needs three tables: a location dimension, a date dimension, and a fact table that holds the actual measurements. You will create all three inside a new analytics schema to keep them separate from the raw staging data.

  • In VS Code, create a new file called transform.py in your project folder.
  • Add the imports, database config, and the first part of your schema creation function by pasting this code:
import psycopg2


DB_CONFIG = {
    "dbname": "weather_pipeline",
    "user": "postgres",
    "password": "your-password-here",
    "host": "localhost",
    "port": "5432",
}


# Creates the analytics schema and all star schema tables
def create_star_schema(cursor):
    cursor.execute("CREATE SCHEMA IF NOT EXISTS analytics;")

    # Location dimension - one row per city
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS analytics.dim_location (
            location_id SERIAL PRIMARY KEY,
            city_name VARCHAR(100) UNIQUE,
            latitude DECIMAL(8, 4),
            longitude DECIMAL(8, 4)
        );
    """)

What does this code do?

The DB_CONFIG dictionary holds the same connection details you used in load.py. Replace your-password-here with your actual PostgreSQL password.

The create_star_schema function first creates an analytics schema (a namespace separate from staging). Then it creates dim_location with a UNIQUE constraint on city_name so each city appears exactly once.

  • Continue building the create_star_schema function by adding the date dimension and fact table directly below the code you just pasted:
    # Date dimension - one row per unique hour
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS analytics.dim_date (
            date_id SERIAL PRIMARY KEY,
            full_timestamp TIMESTAMP UNIQUE,
            full_date DATE,
            year INTEGER,
            month INTEGER,
            day INTEGER,
            day_of_week VARCHAR(10),
            hour INTEGER
        );
    """)

    # Fact table - weather measurements with foreign keys
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS analytics.fact_weather (
            weather_id SERIAL PRIMARY KEY,
            location_id INTEGER REFERENCES analytics.dim_location(location_id),
            date_id INTEGER REFERENCES analytics.dim_date(date_id),
            temperature_c DECIMAL(5, 2),
            relative_humidity INTEGER,
            precipitation_mm DECIMAL(6, 2),
            wind_speed_kmh DECIMAL(6, 2)
        );
    """)

What does dim_date and fact_weather give you?

The dim_date table breaks each timestamp into queryable parts: year, month, day, day of week, and hour. Instead of parsing strings every time you query, you just join to this dimension and filter by any component.

The fact_weather table stores only measurement values (temperature, humidity, precipitation, wind speed) plus foreign keys pointing to the relevant location and date rows. No repeated city names, no repeated coordinates.

  • Save transform.py.
  • Confirm it has no syntax errors by running this command in your terminal:
python -c "import transform; print('Schema function loaded')"

You should see Schema function loaded printed with no errors.

Seeing a SyntaxError?

Check that the triple-quoted SQL strings (""") are closed properly. Each cursor.execute(""" must have a matching """) on a later line. Also confirm the indentation inside create_star_schema is consistent (4 spaces per level).

help me fix a SyntaxError in my transform.py star schema function

Populate dimension tables

With the schema structure defined, you need functions to fill the dimension tables from your staging data. Each dimension pulls distinct values so there is no duplication.

  • Add the populate_dim_location function below create_star_schema in transform.py:

def populate_dim_location(cursor):
    # Pull unique cities from staging into the location dimension
    cursor.execute("""
        INSERT INTO analytics.dim_location (city_name, latitude, longitude)
        SELECT DISTINCT city_name, latitude, longitude
        FROM staging.raw_weather
        ON CONFLICT (city_name) DO NOTHING;
    """)
    cursor.execute("SELECT COUNT(*) FROM analytics.dim_location;")
    count = cursor.fetchone()[0]
    print(f"dim_location: {count} rows")

What does ON CONFLICT DO NOTHING do?

Because city_name has a UNIQUE constraint, inserting a duplicate city would normally throw an error. ON CONFLICT DO NOTHING tells PostgreSQL to skip any row that would violate the constraint. This makes the function safe to run multiple times without creating duplicates.

  • Add the populate_dim_date function directly below populate_dim_location:

def populate_dim_date(cursor):
    # Parse each unique timestamp into date components
    cursor.execute("""
        INSERT INTO analytics.dim_date (full_timestamp, full_date, year, month, day, day_of_week, hour)
        SELECT DISTINCT
            timestamp::TIMESTAMP AS full_timestamp,
            timestamp::DATE AS full_date,
            EXTRACT(YEAR FROM timestamp::TIMESTAMP)::INTEGER AS year,
            EXTRACT(MONTH FROM timestamp::TIMESTAMP)::INTEGER AS month,
            EXTRACT(DAY FROM timestamp::TIMESTAMP)::INTEGER AS day,
            TO_CHAR(timestamp::TIMESTAMP, 'Day') AS day_of_week,
            EXTRACT(HOUR FROM timestamp::TIMESTAMP)::INTEGER AS hour
        FROM staging.raw_weather
        ON CONFLICT (full_timestamp) DO NOTHING;
    """)
    cursor.execute("SELECT COUNT(*) FROM analytics.dim_date;")
    count = cursor.fetchone()[0]
    print(f"dim_date: {count} rows")

What does this SQL do?

The timestamp::TIMESTAMP cast converts the raw string from staging into a proper PostgreSQL timestamp. Then EXTRACT pulls each date part (year, month, day, hour) as an integer.

The TO_CHAR(timestamp::TIMESTAMP, 'Day') function returns the day name (Monday, Tuesday, etc.) as a readable string. With 168 unique hourly timestamps across 7 days, this dimension will hold exactly 168 rows.

  • Save transform.py.
  • Confirm both new functions load without errors by running:
python -c "import transform; print('All functions loaded')"

You should see All functions loaded with no traceback.

Seeing an IndentationError?

Each function definition (def populate_dim_location, def populate_dim_date) must start at column 0 (no indentation). The code inside each function must be indented by 4 spaces. Make sure you did not accidentally paste the function inside create_star_schema.

help me fix an IndentationError in transform.py

Populate the fact table and run the transformation

The fact table is the heart of the star schema. It joins the staging data to both dimension tables and stores only the measurement values with foreign key references.

  • Add the populate_fact_weather function below populate_dim_date in transform.py:

def populate_fact_weather(cursor):
    # Clear existing facts for a clean reload
    cursor.execute("TRUNCATE TABLE analytics.fact_weather RESTART IDENTITY;")
    # Join staging to both dimensions and insert measurements
    cursor.execute("""
        INSERT INTO analytics.fact_weather
            (location_id, date_id, temperature_c, relative_humidity, precipitation_mm, wind_speed_kmh)
        SELECT
            dl.location_id,
            dd.date_id,
            rw.temperature_c,
            rw.relative_humidity,
            rw.precipitation_mm,
            rw.wind_speed_kmh
        FROM staging.raw_weather rw
        JOIN analytics.dim_location dl ON rw.city_name = dl.city_name
        JOIN analytics.dim_date dd ON rw.timestamp::TIMESTAMP = dd.full_timestamp;
    """)
    cursor.execute("SELECT COUNT(*) FROM analytics.fact_weather;")
    count = cursor.fetchone()[0]
    print(f"fact_weather: {count} rows")

How does the fact table join work?

The JOIN on rw.city_name = dl.city_name looks up the location_id for each city. The join on rw.timestamp::TIMESTAMP = dd.full_timestamp looks up the date_id for each hour.

The result: each fact row stores only the numeric measurements plus two integer keys. No repeated city names. No repeated coordinates. That is what makes star schema queries fast and clean.

  • Finally, add the orchestrator function and the main entry point at the bottom of transform.py:

def transform():
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()

    # Run each transformation step in order
    create_star_schema(cursor)
    populate_dim_location(cursor)
    populate_dim_date(cursor)
    populate_fact_weather(cursor)

    conn.commit()
    cursor.close()
    conn.close()
    print("\nStar schema transformation complete!")


if __name__ == "__main__":
    transform()

What does the transform function do?

The transform() function opens a single database connection, then calls each step in sequence: create the tables, populate locations, populate dates, populate facts. Only after all four succeed does it call conn.commit() to save everything in one transaction.

  • Save transform.py.

Before you run the transformation, what do you think the output will show? How many rows should appear in each table given 5 cities and 168 hourly timestamps?

  • Run the transformation by executing:
python transform.py

You should see output like this:

dim_location: 5 rows
dim_date: 168 rows
fact_weather: 840 rows

Star schema transformation complete!

5 locations, 168 unique hours, and 840 fact rows (5 cities times 168 hours). Every number checks out.

  • Switch to pgAdmin and open the Query Tool on your weather_pipeline database.
  • Run this query to confirm the dimension table:
SELECT * FROM analytics.dim_location;

You should see exactly 5 rows: London, New York, Tokyo, Sydney, and Berlin, each with their latitude and longitude.

  • Verify the fact table row count by running:
SELECT COUNT(*) FROM analytics.fact_weather;

You should see 840 in the result.

Seeing 0 rows or a relation does not exist error?

If analytics.fact_weather does not exist, the script may have failed silently. Re-run python transform.py and look for error messages. Common causes: wrong password in DB_CONFIG, or the staging.raw_weather table is empty (re-run python load.py first).

help me debug why my transform.py produced 0 rows in the analytics tables

✔️ Awesome, I've got everything!

Great. Double check you have saved transform.py and that all five functions (create_star_schema, populate_dim_location, populate_dim_date, populate_fact_weather, transform) are present.

ⓧ I'd like to double check the full code

Your transform.py should contain these sections in order. Scroll up and confirm each is present:

  • The import psycopg2 line and DB_CONFIG dictionary at the top.
  • The create_star_schema(cursor) function creating the analytics schema, dim_location, dim_date, and fact_weather tables.
  • The populate_dim_location(cursor) function using SELECT DISTINCT with ON CONFLICT DO NOTHING.
  • The populate_dim_date(cursor) function using EXTRACT and TO_CHAR to parse timestamps.
  • The populate_fact_weather(cursor) function with JOIN on both dimensions.
  • The transform() orchestrator calling all four functions, plus the if __name__ == "__main__" block at the bottom.

Your staging table held 840 identical-looking rows with city data repeated everywhere. Now you have a clean star schema: 5 location rows, 168 date rows, and 840 fact rows linking them together. Next up, you will write analytical queries that prove just how much easier this structure makes real data analysis.

Run Analytical Queries

Your star schema is built and populated. You have clean dimension tables separating location and date data from the raw measurements, and a lean fact table pointing to both.

Now it is time to prove the schema earns its keep. You will write analytical queries that would have been painful against the flat staging table. Window functions, CTEs, and aggregations become clean and readable when facts and dimensions are properly separated.

In this step, get ready to:
  • Rank cities by temperature each hour using a window function.
  • Calculate daily temperature swings using a CTE and filter for extreme days.
  • Compute a 6-hour rolling precipitation average per city.
Rank cities by temperature using a window function

The RANK() window function assigns a ranking position to each row within a group. You will use it to find which city is hottest and coldest for each hour of the current day.

  • Switch to pgAdmin in your browser.
  • In the left sidebar, expand Servers > your local server > Databases > weather_pipeline.
  • Click Tools in the top menu bar, then click Query Tool.
  • Paste the following query into the Query Tool editor:
SELECT
    dd.hour,
    dl.city_name,
    fw.temperature_c,
    RANK() OVER (
        PARTITION BY dd.hour
        ORDER BY fw.temperature_c DESC
    ) AS temp_rank
FROM analytics.fact_weather fw
JOIN analytics.dim_location dl
    ON fw.location_id = dl.location_id
JOIN analytics.dim_date dd
    ON fw.date_id = dd.date_id
WHERE dd.full_date = CURRENT_DATE
ORDER BY dd.hour, temp_rank;

What does this query do?

PARTITION BY dd.hour groups the rows by hour. Within each hourly group, ORDER BY fw.temperature_c DESC sorts cities from hottest to coldest.

RANK() then assigns position 1 to the hottest city, position 2 to the next, and so on. The result is a leaderboard for every hour of the day.

  • Click the Execute/Run button (the play icon) or press F5.

You should see a results grid with columns for hour, city_name, temperature_c, and temp_rank. Each hour shows 5 rows (one per city), ranked from hottest to coldest.

Seeing zero rows?

The WHERE dd.full_date = CURRENT_DATE filter checks today's date. If you extracted data yesterday and are querying today past the forecast window, try replacing CURRENT_DATE with a date you know is in your data. Run SELECT DISTINCT full_date FROM analytics.dim_date; to see available dates.

Need more help?

Calculate daily temperature swings with a CTE

A CTE (Common Table Expression) lets you define a temporary result set and then query it. You will use one to calculate each city's daily temperature swing, then filter to show only days with a swing greater than 5 degrees Celsius.

  • Clear the Query Tool editor (select all and delete), then paste this query:
WITH daily_stats AS (
    SELECT
        dl.city_name,
        dd.full_date,
        MIN(fw.temperature_c) AS min_temp,
        MAX(fw.temperature_c) AS max_temp,
        AVG(fw.temperature_c) AS avg_temp
    FROM analytics.fact_weather fw
    JOIN analytics.dim_location dl
        ON fw.location_id = dl.location_id
    JOIN analytics.dim_date dd
        ON fw.date_id = dd.date_id
    GROUP BY dl.city_name, dd.full_date
)
SELECT
    city_name,
    full_date,
    min_temp,
    max_temp,
    avg_temp,
    (max_temp - min_temp) AS temp_swing
FROM daily_stats
WHERE (max_temp - min_temp) > 5
ORDER BY temp_swing DESC;

What does this query do?

The WITH daily_stats AS (...) block calculates min, max, and average temperature per city per day. This temporary result set is then available to query as if it were a real table.

The outer SELECT filters to only show city-days where the temperature swing exceeds 5 degrees. Cities with stable temperatures are excluded entirely.

  • Click Execute/Run or press F5.

You should see results showing city names, dates, and temperature swings. The rows are sorted by temp_swing DESC, so the most volatile city-day appears first.

Why CTEs over subqueries?

You could achieve the same result with a nested subquery in the FROM clause. CTEs are preferred in analytics because they read top-to-bottom. Each step is named and self-contained, making the query easier to debug and explain in an interview.

Seeing zero rows again?

If no city-day has a swing greater than 5 degrees, your weather data might be unusually stable. Try lowering the threshold from 5 to 3 in the WHERE clause to confirm the query logic works.

Help me troubleshoot my CTE query.

Compute a rolling precipitation average with a window frame

A rolling average smooths out spiky data to reveal trends. You will calculate a 6-hour rolling average of precipitation per city, which helps identify sustained wet or dry periods that a single hourly reading would miss.

  • Clear the Query Tool editor, then paste this query:
SELECT
    dl.city_name,
    dd.full_timestamp,
    fw.precipitation_mm,
    AVG(fw.precipitation_mm) OVER (
        PARTITION BY dl.city_name
        ORDER BY dd.full_timestamp
        ROWS BETWEEN 5 PRECEDING AND CURRENT ROW
    ) AS rolling_avg_precipitation
FROM analytics.fact_weather fw
JOIN analytics.dim_location dl
    ON fw.location_id = dl.location_id
JOIN analytics.dim_date dd
    ON fw.date_id = dd.date_id
ORDER BY dl.city_name, dd.full_timestamp;

What does the window frame do?

ROWS BETWEEN 5 PRECEDING AND CURRENT ROW tells PostgreSQL to average the current row plus the 5 rows before it. Since each row is one hour, this creates a 6-hour rolling window.

PARTITION BY dl.city_name ensures the window resets for each city. London's rolling average never bleeds into Tokyo's data.

Before you run this, what do you expect the rolling_avg_precipitation column to look like compared to the raw precipitation_mm column? Will it be smoother, spikier, or the same?

  • Click Execute/Run or press F5.

The rolling average column is smoother than the raw values. Individual spikes are dampened because each value is averaged with its 5 predecessors. You should see 840 rows total, grouped by city in chronological order.

Results look identical to raw values?

If your cities received zero precipitation across all hours, the rolling average of zero is still zero. Check a few rows for London or Tokyo. If all precipitation_mm values are 0.00, the query is working correctly. The smoothing effect only becomes visible when there is variation in the source data.

Help me understand my rolling average results.

You have proven that your star schema delivers on its promise. Three queries that would have required messy self-joins and string parsing against the flat staging table now read cleanly with proper dimension joins. Your pipeline is complete from extraction to insight.

Ready for a bonus challenge? In the Secret Mission, you will turn these queries into a live Streamlit dashboard with interactive charts and city selectors.

Secret mission

Build a Streamlit Dashboard

Your pipeline works, but it lives in the terminal. What if you could show off your weather data in a live, interactive dashboard? In this secret mission, you will install Streamlit and build a browser-based dashboard that queries your star schema and displays temperature charts, precipitation trends, and real-time metrics for any city you select.

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 PostgreSQL database uses negligible disk space on your machine.

Resources you used:

  • PostgreSQL database weather_pipeline on your local server.
  • Python virtual environment with installed packages (requests, psycopg2-binary, streamlit, pandas).
  • Project files (extract.py, load.py, transform.py, test_connection.py, dashboard.py).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the pipeline with new cities or data sources.

  • The weather_pipeline database stays on your local PostgreSQL server. It uses negligible disk space and costs nothing.
  • Your Python virtual environment remains in your project folder. It takes up a small amount of disk space.
  • You can re-run the pipeline at any time to refresh the data with the latest weather forecasts.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and database stay intact for when you return.

  • If you have the Streamlit dashboard running, press Ctrl+C in the terminal to stop it.
  • Your database and project files remain on disk. You can pick up exactly where you left off by reactivating the virtual environment and running your scripts again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your database and all project files permanently.

  • Drop the database in pgAdmin. Right-click weather_pipeline in the left sidebar under Databases, select Delete/Drop, and confirm.
  • Delete your project folder from the Desktop. On macOS or Linux, run the following command in your terminal:
rm -rf ~/Desktop/weather-pipeline

On Windows, run this in PowerShell:

Remove-Item -Recurse -Force ~\Desktop\weather-pipeline

This removes your virtual environment, all Python scripts, and any other files inside the project folder.

Nice Work!

Nice Work!

You just built a complete data engineering pipeline from scratch. Real weather data flows from a live API into your PostgreSQL database, gets transformed into a proper star schema, and answers analytical questions through clean SQL.

You learned how to:

  • Built a complete ETL pipeline: extraction from a live REST API, loading into a staging table, and transformation into an analytics-ready star schema.
  • Designed and implemented a dimensional model with fact and dimension tables, the same pattern used in production data warehouses.
  • Wrote analytical SQL using window functions, CTEs, and aggregations to surface insights from your pipeline output.
  • Created a live Streamlit dashboard (Secret Mission) turning your backend pipeline into a demo-ready portfolio piece.

Ready to quiz yourself?