Weather Data Pipeline with Airflow and dbt

Build an end-to-end ELT pipeline with Airflow, dbt, and PostgreSQL.

Introduction

30 Second Summary

Weather data powers decisions from "should I bring an umbrella?" to managing entire power grids. Behind every weather app and forecast dashboard sits a data pipeline that extracts, transforms, and delivers fresh data on a schedule.

In this project, you will build a complete ELT pipeline that pulls live weather forecasts from an API, loads them into a database, and transforms them into an analytics-ready star schema. The entire stack runs locally using Docker.

What You'll Build

A production-style weather data platform that ingests hourly forecasts from the Open-Meteo API for 5 cities, transforms raw JSON into dimensional models, and orchestrates everything with Apache Airflow.

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

  • A Docker Compose environment running Airflow, PostgreSQL, and dbt together as a multi-service deployment.
  • An Airflow DAG that schedules and monitors extraction, transformation, and testing tasks with proper retry logic.
  • A dbt project with staging models, dimension tables, and a fact table following star schema design, plus data quality tests that validate the pipeline output.
  • Secret Mission: A daily weather summary mart with anomaly detection that flags unusual temperature swings and heavy precipitation.

Are there any prerequisites?

You should be comfortable with basic Python syntax, basic SQL queries, and navigating the command line. No prior experience with Airflow, dbt, or Docker is required.

Before We Start

Before We Start...

Set Up the Project Structure

Every data pipeline needs infrastructure before you can write a single line of extraction or transformation code. You need databases to store data, an orchestrator to schedule tasks, and a consistent environment that works the same on every machine.

In this step, you'll create the project folder, configure Docker Compose to run Apache Airflow and PostgreSQL together, and verify that all services start successfully.

In this step, get ready to:
  • Create the project folder with subdirectories for DAGs, scripts, and dbt models.
  • Write Docker Compose and Dockerfile configurations for Airflow and PostgreSQL.
  • Start all services and confirm the Airflow UI loads at localhost:8080.
Create the project folders

Your pipeline needs a clear folder structure to keep extraction scripts, DAG definitions, and dbt models separate.

  • Open a terminal in VS Code by pressing Ctrl+` (Windows) or Cmd+` (macOS).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create the project folder and its subdirectories by running these commands:
mkdir weather-data-pipeline
cd weather-data-pipeline
mkdir dags scripts dbt_project logs
  • Confirm the folders exist by running:
ls

You should see dags, scripts, dbt_project, and logs listed in the output.

  • Open the project folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Select the weather-data-pipeline folder on your Desktop, then click Open.

Why open the folder this way?

Opening the project as a folder (not individual files) lets VS Code's Explorer sidebar show your full directory tree. This makes navigating between files much faster as the project grows.

Write the Docker configuration files

Your pipeline runs inside Docker containers. You need three configuration files: a docker-compose.yaml to define all services, a Dockerfile to customize the Airflow image, and an init_db.sql script to set up the data warehouse database.

  • In VS Code's Explorer sidebar, click the new file icon at the top of the file tree.
  • Name the file docker-compose.yaml and press Enter.
  • Paste the following code into docker-compose.yaml:
version: '3.8'

x-airflow-common: &airflow-common
  build:
    context: .
    dockerfile: Dockerfile
  environment: &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
    WAREHOUSE_HOST: postgres
    WAREHOUSE_PORT: '5432'
    WAREHOUSE_DB: weather_dw
    WAREHOUSE_USER: dbt_user
    WAREHOUSE_PASSWORD: dbt_password
  volumes:
    - ./dags:/opt/airflow/dags
    - ./scripts:/opt/airflow/scripts
    - ./dbt_project:/opt/airflow/dbt_project
    - ./logs:/opt/airflow/logs
  depends_on:
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    ports:
      - "5432:5432"
    volumes:
      - postgres-data:/var/lib/postgresql/data
      - ./scripts/init_db.sql:/docker-entrypoint-initdb.d/init_db.sql
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5
      start_period: 5s

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    command:
      - -c
      - |
        airflow db migrate
        airflow users create \
          --username admin \
          --firstname Admin \
          --lastname User \
          --role Admin \
          --email admin@example.com \
          --password admin
    depends_on:
      postgres:
        condition: service_healthy

  airflow-webserver:
    <<: *airflow-common
    command: airflow webserver --port 8080
    ports:
      - "8080:8080"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-scheduler:
    <<: *airflow-common
    command: airflow scheduler
    healthcheck:
      test: ["CMD", "airflow", "jobs", "check", "--job-type", "SchedulerJob", "--hostname", "$(hostname)"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: airflow triggerer
    healthcheck:
      test: ["CMD-SHELL", "airflow jobs check --job-type TriggererJob --hostname \"$(hostname)\""]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

volumes:
  postgres-data:

What does this configuration do?

This file defines five services that work together:

  • postgres runs a PostgreSQL 16 database that stores both Airflow's metadata and your weather data warehouse.
  • airflow-init runs database migrations and creates the admin user on first startup.
  • airflow-webserver serves the Airflow UI on port 8080.
  • airflow-scheduler triggers your DAG runs on schedule.
  • airflow-triggerer handles deferred tasks efficiently.

The x-airflow-common anchor at the top avoids repeating the same build, environment, and volume configuration across all Airflow services. The WAREHOUSE_* environment variables tell your extraction script where to connect for the data warehouse.

Why LocalExecutor instead of CeleryExecutor?

The official Airflow Docker quick-start uses CeleryExecutor, which requires additional Redis and Celery worker containers. LocalExecutor runs tasks directly on the scheduler process. This keeps memory usage lower and avoids two extra containers that add no value for a single-machine development project.

  • Save docker-compose.yaml by pressing Ctrl+S (Windows) or Cmd+S (macOS).

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

version: '3.8'

x-airflow-common: &airflow-common
  build:
    context: .
    dockerfile: Dockerfile
  environment: &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
    WAREHOUSE_HOST: postgres
    WAREHOUSE_PORT: '5432'
    WAREHOUSE_DB: weather_dw
    WAREHOUSE_USER: dbt_user
    WAREHOUSE_PASSWORD: dbt_password
  volumes:
    - ./dags:/opt/airflow/dags
    - ./scripts:/opt/airflow/scripts
    - ./dbt_project:/opt/airflow/dbt_project
    - ./logs:/opt/airflow/logs
  depends_on:
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    ports:
      - "5432:5432"
    volumes:
      - postgres-data:/var/lib/postgresql/data
      - ./scripts/init_db.sql:/docker-entrypoint-initdb.d/init_db.sql
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5
      start_period: 5s

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    command:
      - -c
      - |
        airflow db migrate
        airflow users create \
          --username admin \
          --firstname Admin \
          --lastname User \
          --role Admin \
          --email admin@example.com \
          --password admin
    depends_on:
      postgres:
        condition: service_healthy

  airflow-webserver:
    <<: *airflow-common
    command: airflow webserver --port 8080
    ports:
      - "8080:8080"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-scheduler:
    <<: *airflow-common
    command: airflow scheduler
    healthcheck:
      test: ["CMD", "airflow", "jobs", "check", "--job-type", "SchedulerJob", "--hostname", "$(hostname)"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: airflow triggerer
    healthcheck:
      test: ["CMD-SHELL", "airflow jobs check --job-type TriggererJob --hostname \"$(hostname)\""]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    depends_on:
      airflow-init:
        condition: service_completed_successfully

volumes:
  postgres-data:
  • Create another new file in the project root called Dockerfile (no file extension).
  • Paste the following code into Dockerfile:
FROM apache/airflow:3.2.2

USER root
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

USER airflow

RUN pip install --no-cache-dir \
    dbt-core==1.11.11 \
    dbt-postgres==1.10.0 \
    requests==2.32.3 \
    psycopg2-binary==2.9.10

What does this Dockerfile do?

This extends the official Airflow 3.2.2 image with the Python libraries your pipeline needs.

  • dbt-core and dbt-postgres handle SQL transformations inside the warehouse.
  • requests calls the weather API.
  • psycopg2-binary connects Python to PostgreSQL.
  • gcc and libpq-dev are system libraries needed to compile the PostgreSQL driver.
  • Save Dockerfile.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

FROM apache/airflow:3.2.2

USER root
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

USER airflow

RUN pip install --no-cache-dir \
    dbt-core==1.11.11 \
    dbt-postgres==1.10.0 \
    requests==2.32.3 \
    psycopg2-binary==2.9.10
  • Create a new file inside the scripts folder called init_db.sql.
  • Paste the following code into init_db.sql:
-- Create the data warehouse database and user
CREATE DATABASE weather_dw;
CREATE USER dbt_user WITH PASSWORD 'dbt_password';
GRANT ALL PRIVILEGES ON DATABASE weather_dw TO dbt_user;

-- Connect to weather_dw and set up schemas
\c weather_dw;

GRANT ALL ON SCHEMA public TO dbt_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO dbt_user;

CREATE TABLE IF NOT EXISTS raw_weather (
    id SERIAL PRIMARY KEY,
    city_name VARCHAR(100) NOT NULL,
    latitude DOUBLE PRECISION NOT NULL,
    longitude DOUBLE PRECISION NOT NULL,
    extracted_at TIMESTAMP NOT NULL DEFAULT NOW(),
    response_json JSONB NOT NULL
);

What does this SQL script do?

PostgreSQL runs this script automatically when the container starts for the first time. It creates a separate weather_dw database (your data warehouse) alongside the airflow database that Airflow uses for its own metadata.

The dbt_user account gives dbt its own credentials to manage transformations. The raw_weather table is where your extraction script will land raw API responses as JSON.

  • Save init_db.sql.

Files not showing in the Explorer?

Make sure you opened the weather-data-pipeline folder itself (not a parent folder). You should see dags, scripts, dbt_project, and logs at the top level of the sidebar.

help me troubleshoot why my project files aren't visible in VS Code's Explorer sidebar

Build and start the pipeline services

With all configuration files in place, you can now build the custom Docker image and start every service. The first build downloads the Airflow base image and installs your Python packages, so it takes a few minutes.

Check your Docker Desktop memory

Airflow requires at least 4 GB of memory allocated to Docker containers. In Docker Desktop, click the gear icon to open Settings, then navigate to Resources. Confirm the memory slider is set to 4 GB or higher.

  • In the VS Code terminal, make sure you are inside the weather-data-pipeline folder.
  • Build the Docker image and start all services by running this command:
docker compose up -d --build

What does this command do?

The --build flag tells Docker Compose to build your custom Dockerfile before starting containers. The -d flag runs everything in detached mode so you get your terminal back.

Docker Compose handles the startup order automatically. PostgreSQL starts first, then airflow-init runs migrations and creates the admin user, and finally the webserver, scheduler, and triggerer come online.

The first build takes 3-5 minutes while Docker downloads images and installs packages. Subsequent starts are much faster.

  • Check that all services are running by executing:
docker compose ps

You should see postgres, airflow-webserver, airflow-scheduler, and airflow-triggerer all showing a status of running (or healthy). The airflow-init service will show exited (0) because it only runs once.

  • Open your web browser and navigate to:
http://localhost:8080
  • Log in with username admin and password admin.

You should see the Airflow DAGs list page. It will be empty for now since you haven't created any DAGs yet.

Services not starting?

If docker compose ps shows services restarting or unhealthy, check the logs for errors:

docker compose logs --tail=50
  • If you see "out of memory" errors, increase Docker Desktop's memory to at least 4 GB in Settings > Resources.
  • If PostgreSQL fails to start, make sure no other process is using port 5432.
  • If the build fails with a network error, check your internet connection and run docker compose up -d --build again.

help me debug why my Docker Compose services aren't starting for the weather pipeline project

Your infrastructure is live. Next up, you'll write the Python script that pulls real weather data from the Open-Meteo API and loads it into your PostgreSQL warehouse.

Build the Data Extraction Script

Your infrastructure is running, but the pipeline has no data flowing through it yet. The first job of any data pipeline is to pull data from a source and land it somewhere reliable.

In this step, you'll write the "E" and "L" of ELT. A Python script will call a live weather API, then load the raw response into PostgreSQL so dbt can transform it later.

In this step, get ready to:
  • Create a database initialization script that sets up the data warehouse and landing table.
  • Write an extraction function that calls the Open-Meteo API for 5 cities with retry logic.
  • Write a loading function that inserts raw JSON into PostgreSQL.
Create the database initialization script

Before the extraction script can load data, PostgreSQL needs a dedicated database and a landing table to receive it. The docker-compose.yaml already mounts an init script into PostgreSQL's startup directory. You just need to create the actual file.

  • Create a new file called init_db.sql inside your scripts/ folder, then paste the following code:
-- Create the data warehouse database and user
CREATE DATABASE weather_dw;
CREATE USER dbt_user WITH PASSWORD 'dbt_password';
GRANT ALL PRIVILEGES ON DATABASE weather_dw TO dbt_user;

-- Connect to weather_dw and set up schemas
\c weather_dw;

GRANT ALL ON SCHEMA public TO dbt_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO dbt_user;

CREATE TABLE IF NOT EXISTS raw_weather (
    id SERIAL PRIMARY KEY,
    city_name VARCHAR(100) NOT NULL,
    latitude DOUBLE PRECISION NOT NULL,
    longitude DOUBLE PRECISION NOT NULL,
    extracted_at TIMESTAMP NOT NULL DEFAULT NOW(),
    response_json JSONB NOT NULL
);

What does this script do?

This SQL runs automatically the first time PostgreSQL starts with a fresh data volume.

  • CREATE DATABASE weather_dw creates a separate database for your pipeline's data, keeping it isolated from Airflow's metadata.
  • CREATE USER dbt_user creates a dedicated user that dbt and the extraction script use to connect.
  • \c weather_dw switches the connection to the new database so the table is created in the right place.
  • raw_weather is the landing table where raw API responses are stored as JSONB.

PostgreSQL's init scripts only run when the data volume is brand new. Since your containers already started in Step 1 without this file, you need to recreate the volume.

  • Save scripts/init_db.sql.
  • Stop all containers and remove the database volume by running this command in your terminal:
docker compose down -v
  • Rebuild and restart the services by running:
docker compose up --build -d
  • Wait about 30 seconds for all services to initialize, then verify the database was created by running:
docker compose exec postgres psql -U airflow -c "\l" | grep weather_dw
  • You should see a line containing weather_dw in the output, confirming the database exists.

Not seeing weather_dw in the output?

  • Make sure you ran docker compose down -v with the -v flag. Without it, the old volume persists and PostgreSQL skips the init script.
  • Check for SQL syntax errors in init_db.sql by running docker compose logs postgres and looking for ERROR lines.
  • If containers are still starting, wait another 30 seconds and try the psql command again.

Help me debug my init_db.sql setup

Write the weather API extraction function

The Open-Meteo API provides free weather forecast data with no API key required. Your extraction script will call it for 5 major cities, fetching hourly temperature, humidity, precipitation, and wind speed.

  • Create a new file called extract_weather.py inside your scripts/ folder, then paste the following imports and configuration:
import json
import logging
import os
import time

import psycopg2
import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

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

API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m,weather_code"

What does this code do?

  • CITIES defines the 5 cities your pipeline tracks, each with its GPS coordinates for the API.
  • API_BASE_URL points to the Open-Meteo forecast endpoint, which returns JSON data with no authentication.
  • HOURLY_VARIABLES tells the API which weather metrics to include in each response.
  • The logging setup formats messages with timestamps so you can trace pipeline runs later in Airflow's logs.
  • You won't see a change yet. Next, add the database connection and extraction functions directly below HOURLY_VARIABLES.

Production APIs can fail due to network issues or rate limits. The fetch_weather function handles this with exponential backoff retries.

  • Add the get_db_connection and fetch_weather functions below HOURLY_VARIABLES by pasting this code:


def get_db_connection():
    return psycopg2.connect(
        host=os.environ.get("WAREHOUSE_HOST", "postgres"),
        port=os.environ.get("WAREHOUSE_PORT", "5432"),
        dbname=os.environ.get("WAREHOUSE_DB", "weather_dw"),
        user=os.environ.get("WAREHOUSE_USER", "dbt_user"),
        password=os.environ.get("WAREHOUSE_PASSWORD", "dbt_password"),
    )


def fetch_weather(city, max_retries=3):
    params = {
        "latitude": city["latitude"],
        "longitude": city["longitude"],
        "hourly": HOURLY_VARIABLES,
        "forecast_days": 7,
        "timezone": "auto",
    }

    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(API_BASE_URL, params=params, timeout=30)
            response.raise_for_status()
            logger.info(f"Successfully fetched weather data for {city['name']}")
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.warning(f"Attempt {attempt}/{max_retries} failed for {city['name']}: {e}")
            if attempt < max_retries:
                time.sleep(2 ** attempt)
            else:
                logger.error(f"All retries exhausted for {city['name']}")
                raise

What does this code do?

get_db_connection() reads database credentials from environment variables (set in your docker-compose.yaml) and returns a live PostgreSQL connection.

fetch_weather() calls the Open-Meteo API for a single city. If the request fails, it waits exponentially longer between retries (2 seconds, then 4 seconds). After 3 failed attempts, it raises the error so Airflow can mark the task as failed and trigger its own retry logic.

  • You won't see a change yet. The next chunk adds the loading function that uses this connection.
Write the loading function and run the script

With the extraction in place, you need a function to land that raw JSON into the raw_weather table. In the ELT pattern, you load data in its raw form first, then transform it in the warehouse later.

Why load raw JSON instead of parsed columns?

Loading the full API response as JSONB keeps the extraction layer simple and decoupled from the schema. If the API adds new fields tomorrow, your extraction script doesn't break.

The transformation layer (dbt) handles parsing JSON into typed columns. This separation is what makes ELT more maintainable than ETL for analytics pipelines.

  • Add the load_weather function below fetch_weather by pasting this code:


def load_weather(city, data):
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO raw_weather (city_name, latitude, longitude, response_json)
                VALUES (%s, %s, %s, %s)
                """,
                (city["name"], city["latitude"], city["longitude"], json.dumps(data)),
            )
        conn.commit()
        logger.info(f"Loaded weather data for {city['name']} into raw_weather")
    finally:
        conn.close()

What does this code do?

  • load_weather() opens a database connection, inserts the city metadata plus the full JSON response into raw_weather, then commits and closes the connection.
  • The try/finally block ensures the connection closes even if the insert fails, preventing connection leaks.
  • json.dumps(data) serializes the Python dictionary back to a JSON string so PostgreSQL can store it as JSONB.
  • You won't see a change yet. The final piece is the run() function that orchestrates everything.

The run() function ties extraction and loading together for all 5 cities. Airflow will call this function directly as a task.

  • Add the run() function and main guard at the bottom of the file by pasting this code:


def run():
    logger.info("Starting weather data extraction")
    for city in CITIES:
        data = fetch_weather(city)
        load_weather(city, data)
    logger.info("Weather data extraction complete")


if __name__ == "__main__":
    run()

What does this code do?

  • run() loops through each city in CITIES, calls fetch_weather() to get the API data, then passes it to load_weather() to store it.
  • The if __name__ == "__main__" guard lets you run the script directly for testing, while Airflow imports and calls run() programmatically.
  • Save scripts/extract_weather.py.
  • Test the script by running it inside the Airflow container:
docker compose exec airflow-webserver python /opt/airflow/scripts/extract_weather.py
  • You should see log messages for each city confirming successful extraction and loading, ending with Weather data extraction complete.

Seeing a connection error or table not found?

  • If you see a message about relation "raw_weather" does not exist, the init script did not run. Go back and run docker compose down -v followed by docker compose up --build -d to recreate the volume.
  • If you see a network timeout, your machine may be blocking outbound requests. Check that you have internet access and try the command again.
  • If the container is not running, run docker compose ps to check service status and docker compose up -d to restart.

Help me debug my extraction script

✔️ Awesome, I've got everything!

Great. Double check you've saved both scripts/init_db.sql and scripts/extract_weather.py.

ⓧ I'd like to double check the full code

-- Create the data warehouse database and user
CREATE DATABASE weather_dw;
CREATE USER dbt_user WITH PASSWORD 'dbt_password';
GRANT ALL PRIVILEGES ON DATABASE weather_dw TO dbt_user;

-- Connect to weather_dw and set up schemas
\c weather_dw;

GRANT ALL ON SCHEMA public TO dbt_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO dbt_user;

CREATE TABLE IF NOT EXISTS raw_weather (
    id SERIAL PRIMARY KEY,
    city_name VARCHAR(100) NOT NULL,
    latitude DOUBLE PRECISION NOT NULL,
    longitude DOUBLE PRECISION NOT NULL,
    extracted_at TIMESTAMP NOT NULL DEFAULT NOW(),
    response_json JSONB NOT NULL
);
import json
import logging
import os
import time

import psycopg2
import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

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

API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m,weather_code"


def get_db_connection():
    return psycopg2.connect(
        host=os.environ.get("WAREHOUSE_HOST", "postgres"),
        port=os.environ.get("WAREHOUSE_PORT", "5432"),
        dbname=os.environ.get("WAREHOUSE_DB", "weather_dw"),
        user=os.environ.get("WAREHOUSE_USER", "dbt_user"),
        password=os.environ.get("WAREHOUSE_PASSWORD", "dbt_password"),
    )


def fetch_weather(city, max_retries=3):
    params = {
        "latitude": city["latitude"],
        "longitude": city["longitude"],
        "hourly": HOURLY_VARIABLES,
        "forecast_days": 7,
        "timezone": "auto",
    }

    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(API_BASE_URL, params=params, timeout=30)
            response.raise_for_status()
            logger.info(f"Successfully fetched weather data for {city['name']}")
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.warning(f"Attempt {attempt}/{max_retries} failed for {city['name']}: {e}")
            if attempt < max_retries:
                time.sleep(2 ** attempt)
            else:
                logger.error(f"All retries exhausted for {city['name']}")
                raise


def load_weather(city, data):
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO raw_weather (city_name, latitude, longitude, response_json)
                VALUES (%s, %s, %s, %s)
                """,
                (city["name"], city["latitude"], city["longitude"], json.dumps(data)),
            )
        conn.commit()
        logger.info(f"Loaded weather data for {city['name']} into raw_weather")
    finally:
        conn.close()


def run():
    logger.info("Starting weather data extraction")
    for city in CITIES:
        data = fetch_weather(city)
        load_weather(city, data)
    logger.info("Weather data extraction complete")


if __name__ == "__main__":
    run()

Your extraction and loading layer is working. Next up, you'll create an Airflow DAG that schedules this script to run automatically and monitors it for failures.

Create the Airflow DAG

Your extraction script is built and ready to pull live weather data from the Open-Meteo API. Right now, the only way to run it is manually from the command line.

In this step, you'll create an Airflow DAG that orchestrates your extraction script as a scheduled, monitored task. The DAG defines what runs, in what order, and what happens when something fails.

In this step, get ready to:
  • Create a DAG file that defines pipeline tasks with proper dependencies.
  • Configure scheduling, retry logic, and catchup settings.
  • Trigger the pipeline from the Airflow UI and verify data flows into PostgreSQL.
Write the DAG file with imports and default arguments

A DAG (Directed Acyclic Graph) is how Airflow represents a workflow. Each node in the graph is a task, and the edges define the order tasks run in.

  • In VS Code, create a new file called weather_pipeline_dag.py inside the dags folder.
  • Add the imports, path configuration, and default arguments by pasting this code:
from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

import sys
sys.path.insert(0, "/opt/airflow/scripts")

# Import the extraction function from our script
from extract_weather import run as extract_weather_run

# Default arguments applied to all tasks in this DAG
default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "email_on_failure": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

What does this code do?

The sys.path.insert line tells Python where to find your extract_weather.py script inside the Docker container. Without it, Airflow wouldn't know where to import extract_weather_run from.

The default_args dictionary sets behaviour that applies to every task in the DAG. If any task fails, Airflow will retry it 2 times with a 5-minute delay between attempts.

Define the DAG schedule and task dependencies

Now you'll define the DAG itself with its schedule, then create three tasks that run in sequence.

  • Add the following code below the default_args dictionary in weather_pipeline_dag.py:
with DAG(
    dag_id="weather_elt_pipeline",
    default_args=default_args,
    description="ELT pipeline: Extract weather data, transform with dbt",
    schedule="@daily",
    start_date=datetime(2026, 6, 1),
    catchup=False,
    tags=["weather", "elt", "dbt"],
) as dag:

    # Task 1: Extract weather data from API and load into PostgreSQL
    extract_and_load = PythonOperator(
        task_id="extract_and_load_weather",
        python_callable=extract_weather_run,
    )

    # Task 2: Run dbt models to transform raw data
    dbt_run = BashOperator(
        task_id="dbt_run",
        bash_command="cd /opt/airflow/dbt_project && dbt run --profiles-dir .",
    )

    # Task 3: Run dbt tests to validate data quality
    dbt_test = BashOperator(
        task_id="dbt_test",
        bash_command="cd /opt/airflow/dbt_project && dbt test --profiles-dir .",
    )

    # Define task execution order
    extract_and_load >> dbt_run >> dbt_test

What does this code do?

The with DAG(...) block creates a DAG named weather_elt_pipeline that runs on a @daily schedule. Setting catchup=False means Airflow won't try to backfill runs for every day since the start_date.

Inside the DAG, three tasks are defined. extract_and_load uses a PythonOperator to call your extraction function directly. dbt_run and dbt_test use BashOperators to run shell commands.

The >> operator sets the execution order: extract first, then transform, then test. If any upstream task fails, downstream tasks won't run.

  • Save dags/weather_pipeline_dag.py.

Why BashOperator for dbt instead of PythonOperator?

The dbt CLI is designed to be run as a shell command. Using BashOperator lets you pass the exact same commands you'd type in a terminal. This makes debugging easier because you can run the same command manually inside the container.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

import sys
sys.path.insert(0, "/opt/airflow/scripts")

from extract_weather import run as extract_weather_run

default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "email_on_failure": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="weather_elt_pipeline",
    default_args=default_args,
    description="ELT pipeline: Extract weather data, transform with dbt",
    schedule="@daily",
    start_date=datetime(2026, 6, 1),
    catchup=False,
    tags=["weather", "elt", "dbt"],
) as dag:

    extract_and_load = PythonOperator(
        task_id="extract_and_load_weather",
        python_callable=extract_weather_run,
    )

    dbt_run = BashOperator(
        task_id="dbt_run",
        bash_command="cd /opt/airflow/dbt_project && dbt run --profiles-dir .",
    )

    dbt_test = BashOperator(
        task_id="dbt_test",
        bash_command="cd /opt/airflow/dbt_project && dbt test --profiles-dir .",
    )

    extract_and_load >> dbt_run >> dbt_test
Trigger the DAG and verify extraction succeeds

Your DAG file is saved in the dags/ folder, which is mounted into the Airflow containers. The scheduler will pick it up automatically within about 30 seconds.

  • Return to the Airflow UI in your browser at localhost:8080.
  • Wait about 30 seconds, then refresh the page. You should see weather_elt_pipeline appear in the DAGs list.
  • The DAG starts in a paused state (the toggle on the left is grey). Click the toggle to unpause it.
  • Click the DAG name weather_elt_pipeline to open its detail view.
  • Click the play button (triangle icon) in the top right to trigger a manual run.
  • Watch the extract_and_load_weather task. It will turn dark green when it succeeds.

What about the dbt tasks?

The dbt_run and dbt_test tasks will fail on this run because you haven't created the dbt project yet. That's expected. The important thing is that extract_and_load_weather succeeds, proving your extraction script works inside Airflow.

DAG not appearing in the list?

  • The scheduler can take up to 30 seconds to detect new DAG files. Try refreshing the page again after waiting.
  • Check for a red error banner at the top of the Airflow UI. This usually means a syntax error in your DAG file. Compare your code to the full file in the tab above.
  • Make sure your file is saved inside the dags/ folder (not the project root). The filename must end in .py.

Help me troubleshoot why my Airflow DAG isn't appearing in the web UI

extract_and_load_weather task failed?

  • Click the failed task in the graph view, then click Log to see the error details.
  • A connection refused error means the WAREHOUSE_HOST environment variable isn't reaching the task. Check that your docker-compose.yaml has the WAREHOUSE_HOST: postgres line in the environment section.
  • A module not found error for requests or psycopg2 means the Dockerfile didn't build correctly. Rebuild with docker compose build then restart.

Help me debug why my Airflow extraction task is failing

Your pipeline is now orchestrated. Airflow knows how to run your extraction, and it proved it works by loading real weather data into PostgreSQL. Next up, you'll build the dbt transformation layer that turns that raw JSON into a clean dimensional model.

Build the dbt Transformation Layer

Your extraction pipeline is pulling live weather data into PostgreSQL as raw JSON. Now it's time to transform that data into a clean, queryable dimensional model.

dbt (data build tool) lets you write SQL transformations that run inside your database. In this step, you'll build a staging model that parses JSON into typed columns, then structure the data into dimension and fact tables following the star schema pattern.

In this step, get ready to:
  • Initialize a dbt project with configuration files.
  • Create a staging model that parses raw JSON into typed columns.
  • Build dimension and fact tables with data quality tests.
Initialize the dbt project and configure it

A dbt project needs a specific directory structure and configuration files to know where your models live and how to connect to the database.

  • Create the dbt project directory structure by running this command from your project root folder (the one containing docker-compose.yaml):
mkdir -p dbt_project/models/staging dbt_project/models/marts dbt_project/tests dbt_project/macros dbt_project/seeds dbt_project/snapshots dbt_project/analyses
  • Confirm the directories exist by running:
ls dbt_project

You should see folders named models, tests, macros, seeds, snapshots, and analyses listed in the output.

  • Create the main dbt project configuration file by adding the following to dbt_project/dbt_project.yml:
name: 'weather_pipeline'
version: '1.0.0'

profile: 'weather_pipeline'

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

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

models:
  weather_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table

What does this configuration do?

The name and profile fields link this project to the database connection defined in profiles.yml (which you'll create next).

The models section tells dbt to materialize staging models as views (lightweight, always up-to-date) and mart models as tables (persisted for fast queries).

  • Save dbt_project/dbt_project.yml.

YAML indentation issues?

YAML is sensitive to spacing. Make sure you use spaces (not tabs) and that nested keys are indented by exactly 2 spaces. The +materialized keys should be indented under their parent folder names.

Ask for help: My dbt_project.yml has a YAML parsing error, how do I fix the indentation?

  • Create the database connection profile by adding the following to dbt_project/profiles.yml:
weather_pipeline:
  target: dev
  outputs:
    dev:
      type: postgres
      host: "{{ env_var('WAREHOUSE_HOST', 'postgres') }}"
      port: "{{ env_var('WAREHOUSE_PORT', '5432') | int }}"
      user: "{{ env_var('WAREHOUSE_USER', 'dbt_user') }}"
      password: "{{ env_var('WAREHOUSE_PASSWORD', 'dbt_password') }}"
      dbname: "{{ env_var('WAREHOUSE_DB', 'weather_dw') }}"
      schema: public
      threads: 4

Why use environment variables here?

The env_var() function pulls connection details from the environment variables you already defined in docker-compose.yaml (WAREHOUSE_HOST, WAREHOUSE_DB, etc.). This means dbt automatically connects to the right database when running inside your Airflow containers.

  • Save dbt_project/profiles.yml.
  • Create the package dependencies file by adding the following to dbt_project/packages.yml:
packages:
  - package: dbt-labs/dbt_utils
    version: "1.3.0"

What is dbt_utils?

The dbt_utils package provides helper macros like generate_surrogate_key that you'll use to create unique identifiers for your dimension and fact tables.

  • Save dbt_project/packages.yml.
  • Install the dbt packages by running this command:
docker compose exec airflow-webserver bash -c "cd /opt/airflow/dbt_project && dbt deps --profiles-dir ."

You should see output indicating that dbt_utils was successfully installed.

Seeing a profiles directory error?

Make sure profiles.yml is saved inside the dbt_project/ folder (not the project root). The --profiles-dir . flag tells dbt to look for the profile in the current directory.

Ask for help: dbt deps is failing with a profiles directory error inside my Docker container

✔️ Awesome, I've got everything!

Great. Make sure all three YAML files are saved in your dbt_project/ folder.

ⓧ I'd like to double check the full code

Here is dbt_project/dbt_project.yml:

name: 'weather_pipeline'
version: '1.0.0'

profile: 'weather_pipeline'

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

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

models:
  weather_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table

Here is dbt_project/profiles.yml:

weather_pipeline:
  target: dev
  outputs:
    dev:
      type: postgres
      host: "{{ env_var('WAREHOUSE_HOST', 'postgres') }}"
      port: "{{ env_var('WAREHOUSE_PORT', '5432') | int }}"
      user: "{{ env_var('WAREHOUSE_USER', 'dbt_user') }}"
      password: "{{ env_var('WAREHOUSE_PASSWORD', 'dbt_password') }}"
      dbname: "{{ env_var('WAREHOUSE_DB', 'weather_dw') }}"
      schema: public
      threads: 4

Here is dbt_project/packages.yml:

packages:
  - package: dbt-labs/dbt_utils
    version: "1.3.0"
Create the staging model

The staging layer is where raw data gets cleaned and typed. Your raw_weather table stores the API response as a single JSON blob per city. The staging model will unnest that JSON into one row per city per hour with properly typed columns.

  • Create the staging SQL model by adding the following to dbt_project/models/staging/stg_weather_observations.sql:
with raw as (
    select
        id as raw_id,
        city_name,
        latitude,
        longitude,
        extracted_at,
        response_json
    from raw_weather
),

hourly_data as (
    select
        raw_id,
        city_name,
        latitude,
        longitude,
        extracted_at,
        ordinality - 1 as hour_index,
        time_value,
        (response_json->'hourly'->'temperature_2m'->>( ordinality - 1))::double precision as temperature_2m,
        (response_json->'hourly'->'relative_humidity_2m'->>(ordinality - 1))::integer as relative_humidity_2m,
        (response_json->'hourly'->'precipitation'->>(ordinality - 1))::double precision as precipitation,
        (response_json->'hourly'->'wind_speed_10m'->>(ordinality - 1))::double precision as wind_speed_10m,
        (response_json->'hourly'->'weather_code'->>(ordinality - 1))::integer as weather_code
    from raw,
        jsonb_array_elements_text(response_json->'hourly'->'time')
        with ordinality as t(time_value, ordinality)
)

select
    {{ dbt_utils.generate_surrogate_key(['city_name', 'time_value', 'extracted_at']) }} as observation_id,
    city_name,
    latitude,
    longitude,
    time_value::timestamp as observation_timestamp,
    date(time_value::timestamp) as observation_date,
    temperature_2m,
    relative_humidity_2m,
    precipitation,
    wind_speed_10m,
    weather_code,
    extracted_at
from hourly_data

What does this model do?

The raw CTE selects from the raw_weather table. The hourly_data CTE uses PostgreSQL's jsonb_array_elements_text function to explode the JSON time array into individual rows (one per hour).

Each hourly weather metric is extracted from the JSON using array index access (->> operator) and cast to its proper type. The final select generates a unique observation_id using generate_surrogate_key from dbt_utils.

  • Save dbt_project/models/staging/stg_weather_observations.sql.

You won't see a change yet. The next file defines the tests that validate this model's output.

  • Create the staging schema file with tests by adding the following to dbt_project/models/staging/schema.yml:
version: 2

models:
  - name: stg_weather_observations
    description: "Staged weather observations parsed from raw API JSON responses"
    columns:
      - name: observation_id
        description: "Surrogate key for each observation"
        tests:
          - unique
          - not_null
      - name: city_name
        description: "Name of the city"
        tests:
          - not_null
          - accepted_values:
              values: ['London', 'New York', 'Tokyo', 'Sydney', 'Berlin']
      - name: observation_timestamp
        description: "Timestamp of the weather observation"
        tests:
          - not_null
      - name: temperature_2m
        description: "Air temperature at 2 meters above ground in Celsius"
      - name: relative_humidity_2m
        description: "Relative humidity percentage at 2 meters"
      - name: precipitation
        description: "Precipitation in mm for the preceding hour"
      - name: wind_speed_10m
        description: "Wind speed at 10 meters above ground in km/h"

What do these tests enforce?

  • unique on observation_id ensures no duplicate rows slip through.
  • not_null checks that critical columns always have values.
  • accepted_values on city_name catches any unexpected city names that shouldn't be in the data.
  • Save dbt_project/models/staging/schema.yml.

Seeing unexpected YAML errors?

Make sure the accepted_values list uses single quotes around each city name. Double-check that tests: is indented under the column name, not at the same level.

Ask for help: My dbt schema.yml has a parsing error in the accepted_values section

✔️ Awesome, I've got everything!

Great. Double check both files are saved in dbt_project/models/staging/.

ⓧ I'd like to double check the full code

Here is dbt_project/models/staging/stg_weather_observations.sql:

with raw as (
    select
        id as raw_id,
        city_name,
        latitude,
        longitude,
        extracted_at,
        response_json
    from raw_weather
),

hourly_data as (
    select
        raw_id,
        city_name,
        latitude,
        longitude,
        extracted_at,
        ordinality - 1 as hour_index,
        time_value,
        (response_json->'hourly'->'temperature_2m'->>( ordinality - 1))::double precision as temperature_2m,
        (response_json->'hourly'->'relative_humidity_2m'->>(ordinality - 1))::integer as relative_humidity_2m,
        (response_json->'hourly'->'precipitation'->>(ordinality - 1))::double precision as precipitation,
        (response_json->'hourly'->'wind_speed_10m'->>(ordinality - 1))::double precision as wind_speed_10m,
        (response_json->'hourly'->'weather_code'->>(ordinality - 1))::integer as weather_code
    from raw,
        jsonb_array_elements_text(response_json->'hourly'->'time')
        with ordinality as t(time_value, ordinality)
)

select
    {{ dbt_utils.generate_surrogate_key(['city_name', 'time_value', 'extracted_at']) }} as observation_id,
    city_name,
    latitude,
    longitude,
    time_value::timestamp as observation_timestamp,
    date(time_value::timestamp) as observation_date,
    temperature_2m,
    relative_humidity_2m,
    precipitation,
    wind_speed_10m,
    weather_code,
    extracted_at
from hourly_data

Here is dbt_project/models/staging/schema.yml:

version: 2

models:
  - name: stg_weather_observations
    description: "Staged weather observations parsed from raw API JSON responses"
    columns:
      - name: observation_id
        description: "Surrogate key for each observation"
        tests:
          - unique
          - not_null
      - name: city_name
        description: "Name of the city"
        tests:
          - not_null
          - accepted_values:
              values: ['London', 'New York', 'Tokyo', 'Sydney', 'Berlin']
      - name: observation_timestamp
        description: "Timestamp of the weather observation"
        tests:
          - not_null
      - name: temperature_2m
        description: "Air temperature at 2 meters above ground in Celsius"
      - name: relative_humidity_2m
        description: "Relative humidity percentage at 2 meters"
      - name: precipitation
        description: "Precipitation in mm for the preceding hour"
      - name: wind_speed_10m
        description: "Wind speed at 10 meters above ground in km/h"
Build the dimensional mart layer

The mart layer organizes data into a star schema with dimension tables (descriptive attributes) surrounding a central fact table (measurable events). This pattern makes analytical queries fast and intuitive.

  • Create the location dimension by adding the following to dbt_project/models/marts/dim_location.sql:
select distinct
    {{ dbt_utils.generate_surrogate_key(['city_name']) }} as location_id,
    city_name,
    latitude,
    longitude
from {{ ref('stg_weather_observations') }}

What is a dimension table?

Dimension tables hold descriptive attributes that give context to facts. Here, dim_location stores each city's name and coordinates. The location_id surrogate key links this dimension to the fact table.

  • Save dbt_project/models/marts/dim_location.sql.
  • Create the date dimension by adding the following to dbt_project/models/marts/dim_date.sql:
with date_spine as (
    select distinct observation_date as date_day
    from {{ ref('stg_weather_observations') }}
)

select
    {{ dbt_utils.generate_surrogate_key(['date_day']) }} as date_id,
    date_day,
    extract(year from date_day)::integer as year,
    extract(month from date_day)::integer as month,
    extract(day from date_day)::integer as day_of_month,
    extract(dow from date_day)::integer as day_of_week,
    to_char(date_day, 'Day') as day_name,
    to_char(date_day, 'Month') as month_name,
    case when extract(dow from date_day) in (0, 6) then true else false end as is_weekend
from date_spine

What does dim_date add?

A date dimension pre-calculates calendar attributes (year, month, day of week, weekend flag) so that analysts can filter and group by these without recalculating them in every query. The date_spine CTE pulls the distinct dates from your observations.

  • Save dbt_project/models/marts/dim_date.sql.
  • Create the fact table by adding the following to dbt_project/models/marts/fact_weather_observations.sql:
select
    observation_id,
    {{ dbt_utils.generate_surrogate_key(['city_name']) }} as location_id,
    {{ dbt_utils.generate_surrogate_key(['observation_date']) }} as date_id,
    observation_timestamp,
    temperature_2m,
    relative_humidity_2m,
    precipitation,
    wind_speed_10m,
    weather_code,
    extracted_at
from {{ ref('stg_weather_observations') }}

How does the fact table connect to dimensions?

The fact table uses the same generate_surrogate_key calls as the dimension tables. This generates matching location_id and date_id values that you can join on. The weather metrics (temperature, humidity, precipitation, wind speed) are the measurable facts.

  • Save dbt_project/models/marts/fact_weather_observations.sql.
  • Create the marts schema file with tests and relationship validations by adding the following to dbt_project/models/marts/schema.yml:
version: 2

models:
  - name: dim_location
    description: "Dimension table for city locations"
    columns:
      - name: location_id
        description: "Surrogate key for location"
        tests:
          - unique
          - not_null
      - name: city_name
        tests:
          - not_null

  - name: dim_date
    description: "Date dimension with calendar attributes"
    columns:
      - name: date_id
        description: "Surrogate key for date"
        tests:
          - unique
          - not_null
      - name: date_day
        tests:
          - not_null

  - name: fact_weather_observations
    description: "Fact table with hourly weather observations per city"
    columns:
      - name: observation_id
        description: "Surrogate key for each observation"
        tests:
          - unique
          - not_null
      - name: location_id
        description: "Foreign key to dim_location"
        tests:
          - not_null
          - relationships:
              to: ref('dim_location')
              field: location_id
      - name: date_id
        description: "Foreign key to dim_date"
        tests:
          - not_null
          - relationships:
              to: ref('dim_date')
              field: date_id

What do the relationships tests do?

The relationships test checks referential integrity. It verifies that every location_id in the fact table has a matching row in dim_location. Same for date_id and dim_date. This catches broken joins before they reach your dashboards.

  • Save dbt_project/models/marts/schema.yml.

Not sure if your mart files are in the right place?

All three SQL files (dim_location.sql, dim_date.sql, fact_weather_observations.sql) and schema.yml should all be inside dbt_project/models/marts/. If they're in the wrong folder, dbt won't apply the +materialized: table config from dbt_project.yml.

Ask for help: My dbt mart models aren't being picked up when I run dbt run

✔️ Awesome, I've got everything!

Great. Make sure all four files are saved in dbt_project/models/marts/.

ⓧ I'd like to double check the full code

Here is dbt_project/models/marts/dim_location.sql:

select distinct
    {{ dbt_utils.generate_surrogate_key(['city_name']) }} as location_id,
    city_name,
    latitude,
    longitude
from {{ ref('stg_weather_observations') }}

Here is dbt_project/models/marts/dim_date.sql:

with date_spine as (
    select distinct observation_date as date_day
    from {{ ref('stg_weather_observations') }}
)

select
    {{ dbt_utils.generate_surrogate_key(['date_day']) }} as date_id,
    date_day,
    extract(year from date_day)::integer as year,
    extract(month from date_day)::integer as month,
    extract(day from date_day)::integer as day_of_month,
    extract(dow from date_day)::integer as day_of_week,
    to_char(date_day, 'Day') as day_name,
    to_char(date_day, 'Month') as month_name,
    case when extract(dow from date_day) in (0, 6) then true else false end as is_weekend
from date_spine

Here is dbt_project/models/marts/fact_weather_observations.sql:

select
    observation_id,
    {{ dbt_utils.generate_surrogate_key(['city_name']) }} as location_id,
    {{ dbt_utils.generate_surrogate_key(['observation_date']) }} as date_id,
    observation_timestamp,
    temperature_2m,
    relative_humidity_2m,
    precipitation,
    wind_speed_10m,
    weather_code,
    extracted_at
from {{ ref('stg_weather_observations') }}

Here is dbt_project/models/marts/schema.yml:

version: 2

models:
  - name: dim_location
    description: "Dimension table for city locations"
    columns:
      - name: location_id
        description: "Surrogate key for location"
        tests:
          - unique
          - not_null
      - name: city_name
        tests:
          - not_null

  - name: dim_date
    description: "Date dimension with calendar attributes"
    columns:
      - name: date_id
        description: "Surrogate key for date"
        tests:
          - unique
          - not_null
      - name: date_day
        tests:
          - not_null

  - name: fact_weather_observations
    description: "Fact table with hourly weather observations per city"
    columns:
      - name: observation_id
        description: "Surrogate key for each observation"
        tests:
          - unique
          - not_null
      - name: location_id
        description: "Foreign key to dim_location"
        tests:
          - not_null
          - relationships:
              to: ref('dim_location')
              field: location_id
      - name: date_id
        description: "Foreign key to dim_date"
        tests:
          - not_null
          - relationships:
              to: ref('dim_date')
              field: date_id

Your dbt transformation layer is complete. Next up, you'll wire these dbt commands into the Airflow DAG and run the entire pipeline end-to-end.

Orchestrate the Full Pipeline

In the previous step, you built a complete dbt transformation layer with staging models, dimension tables, and a fact table.

Now it's time to run the entire pipeline end-to-end. Your Airflow DAG already has the task chain wired up: extract data from the API, run dbt transformations, then run dbt tests. In this step, you'll trigger that DAG from the Airflow UI and verify the final output in PostgreSQL.

In this step, get ready to:
  • Review how the DAG wires dbt commands into BashOperator tasks.
  • Trigger the full pipeline from the Airflow UI and verify all tasks pass green.
  • Query the fact_weather_observations table with joins to dimension tables.
Review the DAG's dbt integration

Your DAG file already defines three tasks chained together with the >> operator. The key tasks that execute dbt are dbt_run and dbt_test.

  • In VS Code, open dags/weather_pipeline_dag.py and find the two BashOperator definitions:
    dbt_run = BashOperator(
        task_id="dbt_run",
        bash_command="cd /opt/airflow/dbt_project && dbt run --profiles-dir .",
    )

    dbt_test = BashOperator(
        task_id="dbt_test",
        bash_command="cd /opt/airflow/dbt_project && dbt test --profiles-dir .",
    )

What do these tasks do?

Each BashOperator runs a shell command inside the Airflow container. The dbt_run task navigates to your dbt project directory and executes dbt run, which materializes all your models (staging views, dimension tables, fact table).

The dbt_test task runs dbt test, which executes all the data quality checks you defined in your schema.yml files (not_null, unique, accepted_values, relationships).

The --profiles-dir . flag tells dbt to look for profiles.yml in the current directory instead of the default ~/.dbt/ location.

The task dependency chain at the bottom of the file ensures these run in order:

    extract_and_load >> dbt_run >> dbt_test

Why does task order matter?

The >> operator creates a dependency chain. dbt_run only starts after extract_and_load succeeds, and dbt_test only starts after dbt_run succeeds. If any task fails, downstream tasks are skipped automatically.

Trigger the full pipeline from the Airflow UI

With the dbt project complete from the previous step, all three tasks can now succeed in sequence.

  • Switch back to the Airflow UI in your browser at localhost:8080.
  • Click on weather_elt_pipeline in the DAGs list to open the DAG Details page.
  • If the DAG is paused, click the toggle to unpause it.
  • Click the Trigger Dag button in the action area at the top of the page.
  • In the Dag Trigger Window, select Single Run and confirm the trigger.

What should I see?

After triggering, switch to the Grid View tab on the DAG Details page. You'll see a new DAG run appear with three task boxes. Each task transitions from queued (light grey) to running (light green) to success (dark green).

The full run takes 30-60 seconds. Wait until all three tasks (extract_and_load_weather, dbt_run, dbt_test) show as solid green.

Tasks showing red or failed?

  • Click the failed task box in Grid View, then click Log to see the error message.
  • If dbt_run fails with a "package not found" error, the dbt_utils dependency may not be installed. Run docker compose exec airflow-webserver bash -c "cd /opt/airflow/dbt_project && dbt deps --profiles-dir ." from your terminal and re-trigger the DAG.
  • If extract_and_load_weather fails with a connection error, check that your Docker containers have internet access.

Still stuck? Help me debug a failed Airflow DAG task

Query the transformed data in PostgreSQL

The final proof that your pipeline works is querying the fact table and joining it to your dimension tables. This confirms the star schema relationships are intact.

  • Open a PostgreSQL shell inside your Docker container by running this command:
docker compose exec postgres psql -U dbt_user -d weather_dw
  • Run this query to join the fact table with both dimension tables:
SELECT
    f.observation_timestamp,
    l.city_name,
    d.date_day,
    d.day_name,
    f.temperature_2m,
    f.precipitation
FROM fact_weather_observations f
JOIN dim_location l ON f.location_id = l.location_id
JOIN dim_date d ON f.date_id = d.date_id
LIMIT 10;

What should I see?

You should see 10 rows of weather data with city names (London, New York, Tokyo, Sydney, Berlin), dates, day names, temperatures, and precipitation values. The fact that the JOIN returns results proves your surrogate keys in fact_weather_observations correctly match the keys in dim_location and dim_date.

  • Exit the PostgreSQL shell by typing:
\q

Seeing "relation does not exist"?

  • Make sure your dbt_run task completed successfully in the Airflow UI. If dbt did not run, the tables will not exist.
  • Check that you connected to the weather_dw database (not the default airflow database) by using -d weather_dw in the psql command.

Need help? Help me query my dbt models in PostgreSQL

Secret mission

Build a Daily Summary Mart with Anomaly Detection

Your pipeline delivers hourly observations, but stakeholders want daily summaries with automatic alerting. Build an analytics mart that aggregates hourly data into daily statistics per city and flags anomalous weather days where temperature swings or precipitation exceed safe thresholds.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine using Docker. There are no ongoing cloud costs, but the containers consume memory while running.

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

Resources you used:

  • Docker containers (Airflow webserver, scheduler, triggerer, PostgreSQL).
  • Docker volume (postgres-data) storing your PostgreSQL data.
  • Docker images (custom Airflow image, postgres:16).
  • Local project files (weather-data-pipeline folder on your Desktop).

Keep everything running

No action needed. Choose this if you're still actively building or want to keep experimenting with the pipeline.

  • Your Docker containers will continue running and consuming memory.
  • The Airflow UI remains accessible at localhost:8080.
  • Your DAG will continue to trigger on its daily schedule.

Pause - I'll come back to this later

Shut down running containers to free up memory. Your project files and Docker volumes stay intact so you can resume later.

  • In your terminal, make sure you are inside the weather-data-pipeline folder (the one containing docker-compose.yaml).
  • Stop all containers by running:
docker compose down

This stops and removes the containers but keeps your volumes and images intact. When you want to resume, run:

docker compose up -d

Your data in PostgreSQL will still be there because the postgres-data volume persists.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes containers, volumes, images, and local files.

  • In your terminal, make sure you are inside the weather-data-pipeline folder (the one containing docker-compose.yaml).
  • Stop all containers and remove the Docker volume by running:
docker compose down --volumes --rmi all
  • Delete the project folder by running:

macOS

cd ~/Desktop
rm -rf weather-data-pipeline

Windows

cd ~\Desktop
Remove-Item -Recurse -Force weather-data-pipeline
  • Confirm the folder is gone by running:
ls ~/Desktop

You should no longer see weather-data-pipeline in the output.

Nice Work!

Nice Work!

Nice work. You just built a production-grade end-to-end data pipeline from scratch, running entirely on your local machine with Docker Compose.

You've learned how to:

  • Extracted live weather data from the Open-Meteo API using Python with retry logic, and loaded raw JSON into PostgreSQL following the ELT pattern (load first, transform in-warehouse).
  • Transformed raw JSON into a dimensional star schema using dbt with staging models, dimension tables, and a fact table.
  • Orchestrated the full pipeline with Apache Airflow, including DAG scheduling, retries, task dependencies, and data quality testing with dbt's built-in test framework.
  • Built an analytics mart with anomaly detection, proving you can deliver business value from raw data.

Ready to quiz yourself?