Data Warehouse with DuckDB

Build a local data warehouse using the Medallion Architecture and DuckDB.

Introduction

30 Second Summary

Every day, thousands of data points pour in from sensors, APIs, and user interactions. Making sense of that flood requires a system that organizes raw data into layers, each more refined than the last.

In this project, you will build a local data warehouse that ingests real earthquake data from the USGS API and processes it through three layers of increasing quality using the Medallion Architecture.

What You'll Build

A Python ETL pipeline that extracts real-time earthquake data, loads it into DuckDB, transforms it through Bronze, Silver, and Gold layers, and exports each layer to Parquet files.

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

  • A working Python ETL pipeline script that extracts live earthquake data from the USGS API, cleans it, and produces business-ready aggregations.
  • A DuckDB database file containing three layers of increasingly refined data (Bronze, Silver, Gold).
  • Parquet file exports from each layer, demonstrating the columnar format used in production data engineering.
  • Secret Mission: Push your Gold-layer Parquet files to AWS S3 and query them directly from DuckDB.

Are there any prerequisites?

This project assumes you are comfortable with Python basics (functions, loops, dictionaries) and SQL fundamentals (SELECT, WHERE, GROUP BY). No prior data engineering experience is needed.

Before We Start

In this project, you'll build a local data warehouse that processes real earthquake data through the same layered architecture used by production data platforms.

To get started, you need two Python packages: DuckDB for the database engine, and requests for pulling data from the USGS API.

In this step, get ready to:
  • Create a project folder with a Python virtual environment.
  • Install DuckDB and the requests library.
  • Verify DuckDB works by running a test query.
Create a project folder and set up a Python virtual environment

Python virtual environments keep your project's packages isolated from other projects on your machine. This prevents version conflicts between different projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

This project requires Python 3.10 or higher because that is the minimum version DuckDB supports.

  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

You're all set. Move on to the next instruction below.

ⓧ I see an older version

Your system has an older Python that won't work with DuckDB. macOS often ships with an outdated version by default.

  • Download and install Python 3.10 or newer from the official page: https://www.python.org/downloads/.
  • After installing, close and reopen your terminal.
  • Run python3 --version again to confirm you see 3.10 or higher.

macOS may shadow your new install

If you still see the old version after installing, try running python3.12 --version (or whichever version you installed) to confirm it exists. You can use that specific binary for the rest of this project.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download and install Python 3.10 or newer from the official page: https://www.python.org/downloads/.
  • During installation on Windows, check the box that says Add Python to PATH.
  • Close and reopen your terminal after the install finishes.
  • Run python3 --version to confirm Python is now available.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir earthquake-warehouse
cd earthquake-warehouse
  • Create a Python virtual environment by running:
python3 -m venv venv

What is a virtual environment?

A virtual environment is an isolated copy of Python that lives inside your project folder. Any packages you install go into this local copy, not into your system-wide Python.

This means you can have one project using DuckDB 1.5 and another using a different version without conflict.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate
  • Confirm activation by checking that your terminal prompt now starts with (venv).

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try the activate command again.

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

Help me activate my Python virtual environment.

Install DuckDB and requests packages via pip

With your virtual environment active, any packages you install via pip are scoped to this project only.

  • Install both packages by running:
pip install duckdb requests

What are these packages?

DuckDB is an in-process analytical database. It runs entirely inside your Python script with no separate server to manage.

requests is a Python library for making HTTP calls. You'll use it to pull earthquake data from the USGS API.

  • Verify both packages installed successfully by running:
pip show duckdb requests

You should see output listing each package's name, version, and location.

Seeing an error during installation?

Make sure your virtual environment is active (you should see (venv) in your prompt). If pip is not found, try python3 -m pip install duckdb requests instead.

Help me install DuckDB and requests.

Verify DuckDB installation by running a test query

The fastest way to confirm DuckDB works is to connect to an in-memory database and run a simple SQL query from Python.

  • Run this test command to verify DuckDB connects and executes SQL:
python3 -c "import duckdb; con = duckdb.connect(':memory:'); print(con.execute('SELECT 42 AS answer').fetchone())"

What does this command do?

This one-liner imports the duckdb module, creates a temporary in-memory database, runs a SQL query that returns the number 42, and prints the result.

If everything is installed correctly, you'll see (42,) printed in your terminal.

✔️ I see (42,) in my terminal

DuckDB is installed and working. You're ready to start building your data warehouse.

ⓧ I see an error

If you see ModuleNotFoundError, the duckdb package didn't install into your active environment.

  • Confirm your virtual environment is active (look for (venv) in your prompt).
  • Re-run pip install duckdb and check for errors in the output.
  • Try python3 -m pip install duckdb if the regular pip command doesn't resolve.

Help me fix my DuckDB installation.

ⓧ I see a Python version error

DuckDB requires Python 3.10 or higher. If pip install duckdb failed with a version compatibility error, your Python is too old.

  • Go back to the Python version check above and follow the upgrade instructions.
  • After upgrading, delete the venv folder and recreate it with your new Python version.

Your environment is ready. Next up, you'll write a Python function that pulls live earthquake data from the USGS API.

Set Up Your Environment

Before you can build a data warehouse, you need a workspace with the right tools installed. DuckDB is the analytical database engine that powers your warehouse, and the requests library lets Python talk to web APIs.

In this step, you will set up a project folder, install your dependencies, and create the scaffold for your ETL pipeline script.

In this step, get ready to:
  • Create a project folder and Python virtual environment.
  • Install DuckDB and requests via pip.
  • Create the pipeline script scaffold and verify DuckDB connects.
Create a project folder and set up a Python virtual environment

Your pipeline needs a dedicated folder to hold the script, database file, and Parquet exports. You will also create a virtual environment to keep your project's packages isolated.

First, verify your Python version. DuckDB requires Python 3.10 or higher.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

Your Python version is compatible with DuckDB. Continue to the next instruction below.

ⓧ I see an older version

macOS ships with Python 3.9 by default, which is too old for DuckDB. You need to install Python 3.10 or newer.

  • Download the latest Python installer from python.org/downloads.
  • Run the installer and follow the prompts.
  • After installation, close and reopen your terminal.
  • Run python3 --version again to confirm it shows 3.10 or higher.

macOS shadows newer installs

Even after installing a newer Python, macOS may still resolve python3 to the old system version. If this happens, try running the full path (e.g. python3.12 --version) or use a version manager like pyenv.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download the latest Python installer from python.org/downloads.
  • Run the installer and follow the prompts. On Windows, check the box that says Add Python to PATH before clicking Install Now.
  • Close and reopen your terminal.
  • Run python3 --version to confirm it shows 3.10 or higher.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir earthquake-warehouse
cd earthquake-warehouse
  • Create a Python virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment.

macOS

source venv/bin/activate

Windows

venv\Scripts\Activate.ps1

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

Don't see (venv) in your prompt?

On Windows, PowerShell may block scripts by default. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned and try activating again.

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

help me activate my Python virtual environment

Install DuckDB and requests packages via pip

With your virtual environment active, you can now install the two packages your pipeline needs. duckdb is the analytical database engine. requests handles HTTP calls to the USGS earthquake API.

  • Install both packages by running:
pip install duckdb requests

You should see output showing both packages downloading and installing successfully. The last lines should confirm that duckdb and requests are installed.

Seeing a permission error?

If you see a permission denied error, make sure your virtual environment is active (you should see (venv) in your prompt). Never use sudo pip install inside a virtual environment.

help me fix a pip install error in my virtual environment

Verify DuckDB installation by creating the pipeline scaffold

The best way to verify DuckDB works is to create your pipeline script and run it. This scaffold sets up the project structure you will build on throughout the rest of the project.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the earthquake-warehouse folder on your Desktop, then click Open.
  • In the Explorer sidebar on the left, click the new file icon (a page with a plus sign).
  • Name the file pipeline.py and press Enter.
  • Copy the following code into pipeline.py:
import duckdb
import requests
from datetime import datetime, timezone
import os


def main():
    """Run the full ETL pipeline."""
    # Create output directories for each data layer
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    # Connect to (or create) the DuckDB database file
    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

What does this code do?

  • The imports bring in duckdb (your database engine), requests (for API calls), datetime (for timestamps), and os (for creating directories).
  • os.makedirs creates bronze/, silver/, and gold/ folders for your Medallion Architecture layers.
  • duckdb.connect("earthquake_warehouse.duckdb") creates the database file if it does not exist, or opens it if it does.
  • The if __name__ == "__main__" block ensures main() runs only when you execute the file directly.
  • Save pipeline.py.
  • Back in your terminal (make sure you are in the earthquake-warehouse folder with (venv) active), run the pipeline:
python3 pipeline.py

You should see the banner print and the completion message. This confirms DuckDB connected successfully.

What should I see?

Your terminal output should look like this:

==================================================
EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE
==================================================

✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb
  • Confirm the directories and database file were created by running:
ls

You should see bronze, gold, silver, pipeline.py, earthquake_warehouse.duckdb, and the venv folder in the output.

Seeing a ModuleNotFoundError?

This means Python cannot find the duckdb package. Make sure your virtual environment is active ((venv) in your prompt) before running the script.

If you installed packages without the venv active, reactivate it and run pip install duckdb requests again.

help me fix a ModuleNotFoundError for duckdb in my Python project

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that earthquake_warehouse.duckdb appears in your folder.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file at this point:

import duckdb
import requests
from datetime import datetime, timezone
import os


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your environment is ready and your pipeline scaffold is running. Next up, you will write the extraction function that pulls real earthquake data from the USGS API.

Extract Data from the USGS Earthquake API

Your environment is ready with DuckDB and the requests package installed. Now it's time to pull in some real data.

The USGS Earthquake Hazards Program publishes a free GeoJSON feed of every earthquake detected worldwide. This feed updates every minute and requires no API key. In this step, you'll write a Python function that fetches the past 30 days of magnitude 2.5+ earthquakes from this feed and extracts the data your pipeline needs.

In this step, get ready to:
  • Write an extraction function that calls the USGS GeoJSON feed.
  • Understand the GeoJSON response structure (features, properties, geometry).
  • Run the pipeline and verify the earthquake count.
Write the extraction function

Every ETL pipeline starts with extraction. Your function needs to hit the USGS endpoint, check for errors, parse the JSON response, and return the list of earthquake features.

  • In pipeline.py, add the following function above your main() function:
def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]

What does this code do?

  • url points to the USGS feed for M2.5+ earthquakes over the past 30 days.
  • requests.get(url) makes an HTTP GET request to download the earthquake data.
  • response.raise_for_status() throws an error if the request failed (network issue, server error), so problems surface immediately.
  • data["features"] is the list of individual earthquake records from the GeoJSON response.
  • Save pipeline.py.

You won't see a change yet. The next substep hooks this function into your pipeline's main() so you can run it.

Understand the GeoJSON response structure

The USGS feed returns a GeoJSON FeatureCollection. Each earthquake is a "feature" object containing two key sections.

What is inside each feature?

Each feature has a properties object with fields like mag (magnitude), place (location description), time (epoch milliseconds), and 20+ other attributes.

Each feature also has a geometry object with coordinates in the order [longitude, latitude, depth]. This is standard GeoJSON ordering.

Your function returns the entire features array. Later steps will unpack these properties and coordinates into your Bronze table.

Run the pipeline and verify the extraction

Your extract_earthquake_data() function exists, but main() doesn't call it yet. You need to add the extraction step to the pipeline.

  • In pipeline.py, find this line inside main():
    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)
  • Add the following lines directly below that print banner:
    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

What does this addition do?

This prints a progress label (step 1 of 5 in the pipeline) and then calls your extraction function. The result is stored in features, which later steps will feed into the Bronze layer.

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

You should see output showing the extraction count, similar to:

==================================================
EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE
==================================================

[1/5] Extracting data from USGS API...
Extracted 1042 earthquakes from USGS

✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb

Why is my count different?

The USGS feed is live data that updates every minute. Your count will differ from this example because it reflects the actual number of M2.5+ earthquakes recorded in the past 30 days at the moment you run the script.

Seeing a connection error?

  • Make sure you have an internet connection. The function makes a live HTTP request to the USGS servers.
  • Check that your virtual environment is active. If requests is not found, activate your environment and try again.
  • If you see a timeout, wait a moment and run the script again. The USGS feed is free and public but may occasionally be slow.

Help me debug a connection error when calling the USGS API

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and the extraction count printed successfully.

ⓧ I'd like to double check the full code

import duckdb
import requests
from datetime import datetime, timezone
import os


def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your pipeline now pulls live earthquake data from the USGS feed. Next up, you'll load this raw data into the Bronze layer of your data warehouse.

Build the Bronze Layer

Your pipeline can now pull live earthquake data from USGS. The next question is: where does that raw data go?

In the Medallion Architecture, the Bronze layer is the landing zone for raw data. It stores everything exactly as it arrived from the source, with no transformations or cleaning. In this step, you will create a Bronze layer table in DuckDB and load the raw earthquake records into it.

In this step, get ready to:
  • Connect to DuckDB and create a bronze schema.
  • Create the bronze.earthquakes_raw table with all API fields plus ingestion metadata.
  • Insert the raw extracted data and verify the row count.
Connect to DuckDB and create a bronze schema

Your main() function already connects to DuckDB and stores the connection in con. You will pass that connection into a new load_bronze() function that creates a dedicated schema for raw data.

What is a schema in DuckDB?

A schema is a namespace that groups related tables together. DuckDB creates a default schema called main, but you can create additional schemas to organize your data by layer.

In the Medallion Architecture, you use separate schemas (bronze, silver, gold) to clearly separate raw data from cleaned data from aggregated data.

Create the earthquakes_raw table
  • In pipeline.py, add the load_bronze function below extract_earthquake_data() and above main(). Copy the following code:
def load_bronze(con, features):
    """Load raw earthquake data into the Bronze layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    con.execute("DROP TABLE IF EXISTS bronze.earthquakes_raw")
    con.execute("""
        CREATE TABLE bronze.earthquakes_raw (
            id VARCHAR,
            mag DOUBLE,
            place VARCHAR,
            time_epoch BIGINT,
            updated_epoch BIGINT,
            tz INTEGER,
            url VARCHAR,
            detail VARCHAR,
            felt INTEGER,
            cdi DOUBLE,
            mmi DOUBLE,
            alert VARCHAR,
            status VARCHAR,
            tsunami INTEGER,
            sig INTEGER,
            net VARCHAR,
            code VARCHAR,
            ids VARCHAR,
            sources VARCHAR,
            types VARCHAR,
            nst INTEGER,
            dmin DOUBLE,
            rms DOUBLE,
            gap DOUBLE,
            mag_type VARCHAR,
            event_type VARCHAR,
            longitude DOUBLE,
            latitude DOUBLE,
            depth DOUBLE,
            _ingested_at TIMESTAMP,
            _source_url VARCHAR
        )
    """)

What does this code do?

CREATE SCHEMA IF NOT EXISTS bronze creates the bronze namespace if it does not already exist. This makes the function safe to run multiple times.

DROP TABLE IF EXISTS removes the old table before recreating it. This ensures a clean reload every time the pipeline runs.

The CREATE TABLE statement defines 29 columns matching the USGS API fields, plus two metadata columns: _ingested_at (when you loaded the data) and _source_url (where the data came from). These metadata columns are a Bronze layer best practice for tracing data lineage.

Why store raw data with no transformations?

The Bronze layer preserves the original data exactly as received. If you discover a bug in your transformations later, you can always go back to Bronze and re-process. This is the same pattern used in production lakehouses built on Databricks and Delta Lake.

Insert the raw data and verify the row count
  • Still in pipeline.py, add the following code directly below the CREATE TABLE statement (still inside the load_bronze function):
    ingested_at = datetime.now(timezone.utc).isoformat()
    source_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"

    for feature in features:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        con.execute("""
            INSERT INTO bronze.earthquakes_raw VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, [
            feature["id"],
            props.get("mag"),
            props.get("place"),
            props.get("time"),
            props.get("updated"),
            props.get("tz"),
            props.get("url"),
            props.get("detail"),
            props.get("felt"),
            props.get("cdi"),
            props.get("mmi"),
            props.get("alert"),
            props.get("status"),
            props.get("tsunami"),
            props.get("sig"),
            props.get("net"),
            props.get("code"),
            props.get("ids"),
            props.get("sources"),
            props.get("types"),
            props.get("nst"),
            props.get("dmin"),
            props.get("rms"),
            props.get("gap"),
            props.get("magType"),
            props.get("type"),
            coords[0],
            coords[1],
            coords[2],
            ingested_at,
            source_url,
        ])

    count = con.execute("SELECT COUNT(*) FROM bronze.earthquakes_raw").fetchone()[0]
    print(f"Bronze layer: {count} rows loaded")

What does this code do?

ingested_at captures the current UTC timestamp. This records exactly when the data entered your warehouse.

The for loop iterates through every earthquake feature and inserts it using parameterized queries (the ? placeholders). Parameterized queries protect against SQL injection and handle NULL values safely.

props.get() returns None for missing fields instead of raising an error. The USGS API has optional fields that may not exist for every earthquake.

The final two lines query the row count and print it so you can verify the load matched the extraction count.

  • Now update main() to call your new function. Find this line in main():
    features = extract_earthquake_data()

    con.close()
  • Add the following two lines between features = extract_earthquake_data() and con.close(), so it looks like this:
    features = extract_earthquake_data()

    print("\n[2/5] Loading Bronze layer (raw data)...")
    load_bronze(con, features)

    con.close()
  • Save pipeline.py.
  • Run the pipeline by executing this command in your terminal:
python pipeline.py

You should see output showing the extraction count followed by the Bronze layer count. Both numbers should match.

Seeing a row count mismatch or error?

  • If you see a syntax error, check that the load_bronze function is indented correctly and placed between extract_earthquake_data() and main().
  • If you see NameError: name 'load_bronze' is not defined, make sure your function definition appears before main() in the file.
  • If the counts do not match, confirm you are not filtering or skipping any features in the loop.

help me debug my Bronze layer loading issue

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that your terminal output shows matching counts for extraction and Bronze loading.

ⓧ I'd like to double check the full code

import duckdb
import requests
from datetime import datetime, timezone
import os


def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]


def load_bronze(con, features):
    """Load raw earthquake data into the Bronze layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    con.execute("DROP TABLE IF EXISTS bronze.earthquakes_raw")
    con.execute("""
        CREATE TABLE bronze.earthquakes_raw (
            id VARCHAR,
            mag DOUBLE,
            place VARCHAR,
            time_epoch BIGINT,
            updated_epoch BIGINT,
            tz INTEGER,
            url VARCHAR,
            detail VARCHAR,
            felt INTEGER,
            cdi DOUBLE,
            mmi DOUBLE,
            alert VARCHAR,
            status VARCHAR,
            tsunami INTEGER,
            sig INTEGER,
            net VARCHAR,
            code VARCHAR,
            ids VARCHAR,
            sources VARCHAR,
            types VARCHAR,
            nst INTEGER,
            dmin DOUBLE,
            rms DOUBLE,
            gap DOUBLE,
            mag_type VARCHAR,
            event_type VARCHAR,
            longitude DOUBLE,
            latitude DOUBLE,
            depth DOUBLE,
            _ingested_at TIMESTAMP,
            _source_url VARCHAR
        )
    """)

    ingested_at = datetime.now(timezone.utc).isoformat()
    source_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"

    for feature in features:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        con.execute("""
            INSERT INTO bronze.earthquakes_raw VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, [
            feature["id"],
            props.get("mag"),
            props.get("place"),
            props.get("time"),
            props.get("updated"),
            props.get("tz"),
            props.get("url"),
            props.get("detail"),
            props.get("felt"),
            props.get("cdi"),
            props.get("mmi"),
            props.get("alert"),
            props.get("status"),
            props.get("tsunami"),
            props.get("sig"),
            props.get("net"),
            props.get("code"),
            props.get("ids"),
            props.get("sources"),
            props.get("types"),
            props.get("nst"),
            props.get("dmin"),
            props.get("rms"),
            props.get("gap"),
            props.get("magType"),
            props.get("type"),
            coords[0],
            coords[1],
            coords[2],
            ingested_at,
            source_url,
        ])

    count = con.execute("SELECT COUNT(*) FROM bronze.earthquakes_raw").fetchone()[0]
    print(f"Bronze layer: {count} rows loaded")


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

    print("\n[2/5] Loading Bronze layer (raw data)...")
    load_bronze(con, features)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your Bronze layer is loaded with raw earthquake data. Next up, you will transform this raw data into a clean, deduplicated Silver layer ready for analysis.

Build the Silver Layer

Your Bronze layer now holds raw earthquake data exactly as the USGS API delivered it. That's the first piece of the Medallion Architecture puzzle. But raw data is messy.

Timestamps are stored as epoch milliseconds (unreadable numbers like 1716073200000), duplicate records can sneak in when earthquakes get updated, and NULL magnitudes add noise to any analysis. The Silver layer is where you clean all of this up into analysis-ready data.

In this step, get ready to:
  • Create a Silver schema and write a SQL transformation that cleans the Bronze data.
  • Deduplicate earthquake records using SQL window functions.
  • Verify data quality in the Silver layer.
Create the Silver schema and transformation SQL

The Silver layer takes messy Bronze data and transforms it into a consistent, trustworthy format. Your transformation will convert epoch timestamps into proper dates, filter out records with missing magnitudes, and remove duplicates.

  • In pipeline.py, add the following transform_silver function below the load_bronze function:
def transform_silver(con):
    """Transform Bronze data into cleaned Silver layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS silver")
    con.execute("DROP TABLE IF EXISTS silver.earthquakes")
    con.execute("""
        CREATE TABLE silver.earthquakes AS
        WITH deduplicated AS (
            SELECT *,
                ROW_NUMBER() OVER (
                    PARTITION BY id
                    ORDER BY updated_epoch DESC
                ) AS row_num
            FROM bronze.earthquakes_raw
            WHERE mag IS NOT NULL
        )
        SELECT
            id,
            mag,
            place,
            epoch_ms(time_epoch) AS event_time,
            epoch_ms(updated_epoch) AS updated_at,
            url,
            alert,
            status,
            tsunami,
            sig,
            net,
            mag_type,
            event_type,
            longitude,
            latitude,
            depth,
            felt,
            cdi,
            mmi,
            nst,
            dmin,
            rms,
            gap,
            _ingested_at,
            _source_url
        FROM deduplicated
        WHERE row_num = 1
    """)

    count = con.execute("SELECT COUNT(*) FROM silver.earthquakes").fetchone()[0]
    nulls = con.execute("""
        SELECT COUNT(*) FROM silver.earthquakes
        WHERE event_time IS NULL OR mag IS NULL OR id IS NULL
    """).fetchone()[0]
    print(f"Silver layer: {count} rows (removed {nulls} rows with NULL key fields)")

What does this code do?

This function creates a silver schema and builds a cleaned table from the Bronze data using one SQL statement with a CTE (Common Table Expression).

  • WHERE mag IS NOT NULL filters out any earthquake records that have no magnitude reading.
  • epoch_ms(time_epoch) converts raw epoch milliseconds into proper TIMESTAMP values that are human-readable.
  • ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_epoch DESC) assigns a number to each row within groups of the same earthquake ID, ordered by most recent update first.
  • WHERE row_num = 1 keeps only the most recently updated version of each earthquake, eliminating duplicates.
  • The verification queries at the end count total rows and check for NULLs in key columns to confirm data quality.
  • Save pipeline.py.

Seeing a syntax error in the SQL?

Make sure the triple-quoted SQL string (""") opens and closes correctly. The entire SQL from CREATE TABLE through the final WHERE row_num = 1 must be inside one set of triple quotes.

Help me debug my transform_silver function SQL syntax.

Deduplicate records by earthquake ID

The deduplication logic in your SQL is the most important transformation happening in the Silver layer. USGS updates earthquake records as new measurements come in, which can produce multiple rows for the same event.

How does ROW_NUMBER() deduplicate?

ROW_NUMBER() is a window function that numbers rows within groups. PARTITION BY id groups rows by earthquake ID, and ORDER BY updated_epoch DESC puts the most recently updated record first.

After numbering, WHERE row_num = 1 keeps only the top-ranked record for each earthquake. This is a standard deduplication pattern used across production data warehouses.

Now you need to call transform_silver from your main() function so it runs as part of the pipeline.

  • In pipeline.py, scroll down to the main() function.
  • Find this line:
    load_bronze(con, features)

    con.close()
  • Add the following two lines between load_bronze(con, features) and con.close(), so the section looks like this:
    load_bronze(con, features)

    print("\n[3/5] Transforming Silver layer (clean + deduplicate)...")
    transform_silver(con)

    con.close()

What does this change do?

This adds the Silver transformation as step 3 of your pipeline. The print statement logs progress, then transform_silver(con) runs the cleaning and deduplication SQL against your Bronze data.

  • Save pipeline.py.
Create the Silver table and verify data quality

Time to run your pipeline and confirm that the Silver layer produces clean, deduplicated data with no NULLs in key columns.

  • Run your pipeline by executing this command in the terminal:
python pipeline.py
  • Look for the Silver layer output in your terminal. You should see a line like Silver layer: X rows (removed 0 rows with NULL key fields) where X is the number of cleaned records.

What should I see?

Your Silver layer row count should be equal to or slightly less than the Bronze count. Rows are only removed if they had NULL magnitudes or were duplicates. The "removed 0 rows with NULL key fields" message confirms that after transformation, no key columns (event_time, mag, id) contain NULLs.

Seeing an error about bronze.earthquakes_raw not found?

This means the Bronze layer didn't load before the Silver transformation ran. Check that load_bronze(con, features) appears before transform_silver(con) in your main() function.

Help me fix my pipeline so the Silver layer can find Bronze data.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and that your terminal shows the Silver layer row count.

ⓧ I'd like to double check the full code

Here's the full pipeline.py as it should look at the end of this step:

import duckdb
import requests
from datetime import datetime, timezone
import os


def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]


def load_bronze(con, features):
    """Load raw earthquake data into the Bronze layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    con.execute("DROP TABLE IF EXISTS bronze.earthquakes_raw")
    con.execute("""
        CREATE TABLE bronze.earthquakes_raw (
            id VARCHAR,
            mag DOUBLE,
            place VARCHAR,
            time_epoch BIGINT,
            updated_epoch BIGINT,
            tz INTEGER,
            url VARCHAR,
            detail VARCHAR,
            felt INTEGER,
            cdi DOUBLE,
            mmi DOUBLE,
            alert VARCHAR,
            status VARCHAR,
            tsunami INTEGER,
            sig INTEGER,
            net VARCHAR,
            code VARCHAR,
            ids VARCHAR,
            sources VARCHAR,
            types VARCHAR,
            nst INTEGER,
            dmin DOUBLE,
            rms DOUBLE,
            gap DOUBLE,
            mag_type VARCHAR,
            event_type VARCHAR,
            longitude DOUBLE,
            latitude DOUBLE,
            depth DOUBLE,
            _ingested_at TIMESTAMP,
            _source_url VARCHAR
        )
    """)

    ingested_at = datetime.now(timezone.utc).isoformat()
    source_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"

    for feature in features:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        con.execute("""
            INSERT INTO bronze.earthquakes_raw VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, [
            feature["id"],
            props.get("mag"),
            props.get("place"),
            props.get("time"),
            props.get("updated"),
            props.get("tz"),
            props.get("url"),
            props.get("detail"),
            props.get("felt"),
            props.get("cdi"),
            props.get("mmi"),
            props.get("alert"),
            props.get("status"),
            props.get("tsunami"),
            props.get("sig"),
            props.get("net"),
            props.get("code"),
            props.get("ids"),
            props.get("sources"),
            props.get("types"),
            props.get("nst"),
            props.get("dmin"),
            props.get("rms"),
            props.get("gap"),
            props.get("magType"),
            props.get("type"),
            coords[0],
            coords[1],
            coords[2],
            ingested_at,
            source_url,
        ])

    count = con.execute("SELECT COUNT(*) FROM bronze.earthquakes_raw").fetchone()[0]
    print(f"Bronze layer: {count} rows loaded")


def transform_silver(con):
    """Transform Bronze data into cleaned Silver layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS silver")
    con.execute("DROP TABLE IF EXISTS silver.earthquakes")
    con.execute("""
        CREATE TABLE silver.earthquakes AS
        WITH deduplicated AS (
            SELECT *,
                ROW_NUMBER() OVER (
                    PARTITION BY id
                    ORDER BY updated_epoch DESC
                ) AS row_num
            FROM bronze.earthquakes_raw
            WHERE mag IS NOT NULL
        )
        SELECT
            id,
            mag,
            place,
            epoch_ms(time_epoch) AS event_time,
            epoch_ms(updated_epoch) AS updated_at,
            url,
            alert,
            status,
            tsunami,
            sig,
            net,
            mag_type,
            event_type,
            longitude,
            latitude,
            depth,
            felt,
            cdi,
            mmi,
            nst,
            dmin,
            rms,
            gap,
            _ingested_at,
            _source_url
        FROM deduplicated
        WHERE row_num = 1
    """)

    count = con.execute("SELECT COUNT(*) FROM silver.earthquakes").fetchone()[0]
    nulls = con.execute("""
        SELECT COUNT(*) FROM silver.earthquakes
        WHERE event_time IS NULL OR mag IS NULL OR id IS NULL
    """).fetchone()[0]
    print(f"Silver layer: {count} rows (removed {nulls} rows with NULL key fields)")


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

    print("\n[2/5] Loading Bronze layer (raw data)...")
    load_bronze(con, features)

    print("\n[3/5] Transforming Silver layer (clean + deduplicate)...")
    transform_silver(con)

    con.close()
    print("\n\u2713 Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your data warehouse now has two layers working together. Bronze stores the raw truth, and Silver gives you clean, deduplicated data ready for analysis. Next up, you'll build the Gold layer with business-ready aggregations that surface real insights from this earthquake data.

Build the Gold Layer

Your Silver layer now holds clean, deduplicated earthquake records with proper timestamps and separated coordinates. The data is trustworthy, but it is not yet answering questions.

The Gold layer is where your warehouse delivers business value. You will aggregate the Silver data into three purpose-built tables that answer specific questions: how active is seismic activity day by day, how are magnitudes distributed, and which regions experience the most earthquakes.

In this step, get ready to:
  • Create a Gold schema and build a daily earthquake summary table.
  • Build a magnitude distribution table with category buckets.
  • Build a regional activity table that parses location names from the place field.
Create a Gold schema and build the daily summary table

Gold tables are pre-aggregated for specific analytical questions. The first table summarizes earthquake activity per day so an analyst can instantly spot high-activity periods without scanning thousands of rows.

  • In pipeline.py, find the transform_silver function.
  • Below that entire function, add the following code to start your build_gold function:
def build_gold(con):
    """Build Gold layer with business-ready aggregations."""
    con.execute("CREATE SCHEMA IF NOT EXISTS gold")

    # Gold table 1: Daily earthquake summary
    con.execute("DROP TABLE IF EXISTS gold.daily_earthquake_summary")
    con.execute("""
        CREATE TABLE gold.daily_earthquake_summary AS
        SELECT
            CAST(event_time AS DATE) AS event_date,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude,
            ROUND(AVG(depth), 2) AS avg_depth_km
        FROM silver.earthquakes
        GROUP BY CAST(event_time AS DATE)
        ORDER BY event_date DESC
    """)

What does this code do?

This starts the build_gold function by creating the gold schema and your first aggregation table.

  • CAST(event_time AS DATE) strips the time portion so you can group by calendar day.
  • The query calculates count, average magnitude, maximum magnitude, and average depth for each day.
  • ORDER BY event_date DESC puts the most recent days at the top for quick scanning.
  • Save pipeline.py.
  • You won't see output yet. The next two chunks add the remaining Gold tables, then you will run the pipeline.
Build the magnitude distribution table

Seismologists classify earthquakes into standard magnitude bands. This table buckets every earthquake into one of six categories so you can see at a glance whether most activity is minor rumbling or serious events.

  • In pipeline.py, add the following code directly below the daily summary query you just wrote (still inside the build_gold function):
    # Gold table 2: Magnitude distribution
    con.execute("DROP TABLE IF EXISTS gold.magnitude_distribution")
    con.execute("""
        CREATE TABLE gold.magnitude_distribution AS
        SELECT
            CASE
                WHEN mag < 3.0 THEN 'Minor (2.5-2.9)'
                WHEN mag < 4.0 THEN 'Light (3.0-3.9)'
                WHEN mag < 5.0 THEN 'Moderate (4.0-4.9)'
                WHEN mag < 6.0 THEN 'Strong (5.0-5.9)'
                WHEN mag < 7.0 THEN 'Major (6.0-6.9)'
                ELSE 'Great (7.0+)'
            END AS magnitude_category,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(depth), 2) AS avg_depth_km,
            ROUND(AVG(sig), 0) AS avg_significance
        FROM silver.earthquakes
        GROUP BY magnitude_category
        ORDER BY MIN(mag)
    """)

What does the CASE statement do here?

The SQL CASE expression works like a series of if/else conditions. Each earthquake's magnitude is checked against thresholds and assigned to a human-readable category string.

ORDER BY MIN(mag) sorts the output from weakest to strongest category. The avg_significance column uses the USGS sig field, which measures how noteworthy an event is based on magnitude, felt reports, and damage.

  • Save pipeline.py.
  • You won't see output yet. The next chunk adds the final Gold table and the verification print statements.
Build the regional activity table

USGS place names follow a pattern like "10 km NE of Los Angeles, California". The region table parses out the location portion after " of " and aggregates earthquake counts by geographic area.

  • In pipeline.py, add the following code directly below the magnitude distribution query (still inside build_gold):
    # Gold table 3: Regional activity
    con.execute("DROP TABLE IF EXISTS gold.regional_activity")
    con.execute("""
        CREATE TABLE gold.regional_activity AS
        SELECT
            CASE
                WHEN place LIKE '% of %' THEN SPLIT_PART(place, ' of ', 2)
                ELSE place
            END AS region,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude
        FROM silver.earthquakes
        WHERE place IS NOT NULL
        GROUP BY region
        HAVING COUNT(*) >= 3
        ORDER BY earthquake_count DESC
    """)

    for table in ['daily_earthquake_summary', 'magnitude_distribution', 'regional_activity']:
        count = con.execute(f"SELECT COUNT(*) FROM gold.{table}").fetchone()[0]
        print(f"Gold layer: gold.{table} - {count} rows")

What does this code do?

  • SPLIT_PART(place, ' of ', 2) extracts everything after " of " in strings like "10 km NE of Alaska", giving you "Alaska".
  • The CASE handles entries without " of " by using the full place name as the region.
  • HAVING COUNT(*) >= 3 filters out regions with very few events, keeping only areas with meaningful activity.
  • The for loop at the end prints row counts for all three Gold tables as a verification summary.
  • Save pipeline.py.

Now you need to call this new function from your pipeline's main flow.

  • In the main() function, find this line:
    print("\n[3/5] Transforming Silver layer (clean + deduplicate)...")
    transform_silver(con)
  • Add these two lines directly below transform_silver(con), before the con.close() line:
    print("\n[4/5] Building Gold layer (aggregations)...")
    build_gold(con)
  • Save pipeline.py.
  • Run your pipeline by executing this command in your terminal:
python pipeline.py
  • You should see output showing all five pipeline stages, including lines like Gold layer: gold.daily_earthquake_summary - X rows for each of the three Gold tables.

Seeing an error about silver.earthquakes not found?

  • The pipeline runs all stages in order, so Bronze and Silver are always recreated first. If you see this error, make sure your transform_silver function is still intact and runs before build_gold in main().
  • Check that your indentation is correct. All three Gold table blocks must be inside the build_gold function (indented by 4 spaces).
  • Help me debug my Gold layer error.

Why three separate Gold tables instead of one?

Each Gold table answers one business question. This is a core principle of the Medallion Architecture. Analysts and dashboards can query a single focused table without writing complex joins or aggregations themselves.

In production, Gold tables are often refreshed on a schedule and serve as the source for BI tools like Tableau, Looker, or Power BI.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and confirmed the Gold layer row counts printed in your terminal.

ⓧ I'd like to double check the full code

Here is the full pipeline.py at the end of this step:

import duckdb
import requests
from datetime import datetime, timezone
import os


def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]


def load_bronze(con, features):
    """Load raw earthquake data into the Bronze layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    con.execute("DROP TABLE IF EXISTS bronze.earthquakes_raw")
    con.execute("""
        CREATE TABLE bronze.earthquakes_raw (
            id VARCHAR,
            mag DOUBLE,
            place VARCHAR,
            time_epoch BIGINT,
            updated_epoch BIGINT,
            tz INTEGER,
            url VARCHAR,
            detail VARCHAR,
            felt INTEGER,
            cdi DOUBLE,
            mmi DOUBLE,
            alert VARCHAR,
            status VARCHAR,
            tsunami INTEGER,
            sig INTEGER,
            net VARCHAR,
            code VARCHAR,
            ids VARCHAR,
            sources VARCHAR,
            types VARCHAR,
            nst INTEGER,
            dmin DOUBLE,
            rms DOUBLE,
            gap DOUBLE,
            mag_type VARCHAR,
            event_type VARCHAR,
            longitude DOUBLE,
            latitude DOUBLE,
            depth DOUBLE,
            _ingested_at TIMESTAMP,
            _source_url VARCHAR
        )
    """)

    ingested_at = datetime.now(timezone.utc).isoformat()
    source_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"

    for feature in features:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        con.execute("""
            INSERT INTO bronze.earthquakes_raw VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, [
            feature["id"],
            props.get("mag"),
            props.get("place"),
            props.get("time"),
            props.get("updated"),
            props.get("tz"),
            props.get("url"),
            props.get("detail"),
            props.get("felt"),
            props.get("cdi"),
            props.get("mmi"),
            props.get("alert"),
            props.get("status"),
            props.get("tsunami"),
            props.get("sig"),
            props.get("net"),
            props.get("code"),
            props.get("ids"),
            props.get("sources"),
            props.get("types"),
            props.get("nst"),
            props.get("dmin"),
            props.get("rms"),
            props.get("gap"),
            props.get("magType"),
            props.get("type"),
            coords[0],
            coords[1],
            coords[2],
            ingested_at,
            source_url,
        ])

    count = con.execute("SELECT COUNT(*) FROM bronze.earthquakes_raw").fetchone()[0]
    print(f"Bronze layer: {count} rows loaded")


def transform_silver(con):
    """Transform Bronze data into cleaned Silver layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS silver")
    con.execute("DROP TABLE IF EXISTS silver.earthquakes")
    con.execute("""
        CREATE TABLE silver.earthquakes AS
        WITH deduplicated AS (
            SELECT *,
                ROW_NUMBER() OVER (
                    PARTITION BY id
                    ORDER BY updated_epoch DESC
                ) AS row_num
            FROM bronze.earthquakes_raw
            WHERE mag IS NOT NULL
        )
        SELECT
            id,
            mag,
            place,
            epoch_ms(time_epoch) AS event_time,
            epoch_ms(updated_epoch) AS updated_at,
            url,
            alert,
            status,
            tsunami,
            sig,
            net,
            mag_type,
            event_type,
            longitude,
            latitude,
            depth,
            felt,
            cdi,
            mmi,
            nst,
            dmin,
            rms,
            gap,
            _ingested_at,
            _source_url
        FROM deduplicated
        WHERE row_num = 1
    """)

    count = con.execute("SELECT COUNT(*) FROM silver.earthquakes").fetchone()[0]
    nulls = con.execute("""
        SELECT COUNT(*) FROM silver.earthquakes
        WHERE event_time IS NULL OR mag IS NULL OR id IS NULL
    """).fetchone()[0]
    print(f"Silver layer: {count} rows (removed {nulls} rows with NULL key fields)")


def build_gold(con):
    """Build Gold layer with business-ready aggregations."""
    con.execute("CREATE SCHEMA IF NOT EXISTS gold")

    # Gold table 1: Daily earthquake summary
    con.execute("DROP TABLE IF EXISTS gold.daily_earthquake_summary")
    con.execute("""
        CREATE TABLE gold.daily_earthquake_summary AS
        SELECT
            CAST(event_time AS DATE) AS event_date,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude,
            ROUND(AVG(depth), 2) AS avg_depth_km
        FROM silver.earthquakes
        GROUP BY CAST(event_time AS DATE)
        ORDER BY event_date DESC
    """)

    # Gold table 2: Magnitude distribution
    con.execute("DROP TABLE IF EXISTS gold.magnitude_distribution")
    con.execute("""
        CREATE TABLE gold.magnitude_distribution AS
        SELECT
            CASE
                WHEN mag < 3.0 THEN 'Minor (2.5-2.9)'
                WHEN mag < 4.0 THEN 'Light (3.0-3.9)'
                WHEN mag < 5.0 THEN 'Moderate (4.0-4.9)'
                WHEN mag < 6.0 THEN 'Strong (5.0-5.9)'
                WHEN mag < 7.0 THEN 'Major (6.0-6.9)'
                ELSE 'Great (7.0+)'
            END AS magnitude_category,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(depth), 2) AS avg_depth_km,
            ROUND(AVG(sig), 0) AS avg_significance
        FROM silver.earthquakes
        GROUP BY magnitude_category
        ORDER BY MIN(mag)
    """)

    # Gold table 3: Regional activity
    con.execute("DROP TABLE IF EXISTS gold.regional_activity")
    con.execute("""
        CREATE TABLE gold.regional_activity AS
        SELECT
            CASE
                WHEN place LIKE '% of %' THEN SPLIT_PART(place, ' of ', 2)
                ELSE place
            END AS region,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude
        FROM silver.earthquakes
        WHERE place IS NOT NULL
        GROUP BY region
        HAVING COUNT(*) >= 3
        ORDER BY earthquake_count DESC
    """)

    for table in ['daily_earthquake_summary', 'magnitude_distribution', 'regional_activity']:
        count = con.execute(f"SELECT COUNT(*) FROM gold.{table}").fetchone()[0]
        print(f"Gold layer: gold.{table} - {count} rows")


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

    print("\n[2/5] Loading Bronze layer (raw data)...")
    load_bronze(con, features)

    print("\n[3/5] Transforming Silver layer (clean + deduplicate)...")
    transform_silver(con)

    print("\n[4/5] Building Gold layer (aggregations)...")
    build_gold(con)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your warehouse now has three complete layers. Next up, you will export these tables to Parquet files and run analytical queries that surface real insights from the Gold layer.

Export to Parquet and Query

Your data warehouse now has all three layers populated with real earthquake data. The final step in any production pipeline is making that data portable and queryable outside the database itself.

In this step, you'll export each layer to Parquet files. Parquet is the columnar file format that dominates modern data engineering because it compresses well and supports fast analytical reads.

In this step, get ready to:
  • Export all five tables to Parquet files using DuckDB's COPY statement.
  • Query Parquet files directly without importing them back into a table.
  • Run analytical queries against your Gold tables to surface earthquake insights.
Export each layer to Parquet files

DuckDB can write any table directly to a Parquet file with a single COPY statement. You already have the bronze/, silver/, and gold/ directories created by your main() function, so the export paths are ready.

  • In pipeline.py, add the following function below your build_gold(con) function:
def export_to_parquet(con):
    """Export each layer to Parquet files."""
    con.execute("COPY bronze.earthquakes_raw TO 'bronze/earthquakes_raw.parquet' (FORMAT parquet)")
    con.execute("COPY silver.earthquakes TO 'silver/earthquakes.parquet' (FORMAT parquet)")
    con.execute("COPY gold.daily_earthquake_summary TO 'gold/daily_earthquake_summary.parquet' (FORMAT parquet)")
    con.execute("COPY gold.magnitude_distribution TO 'gold/magnitude_distribution.parquet' (FORMAT parquet)")
    con.execute("COPY gold.regional_activity TO 'gold/regional_activity.parquet' (FORMAT parquet)")
    print("Parquet files exported to bronze/, silver/, gold/ directories")

What does this code do?

Each COPY ... TO ... (FORMAT parquet) statement serializes one table into a Parquet file. DuckDB uses snappy compression by default, which balances file size and read speed.

You end up with five Parquet files mirroring your three-layer architecture. These files are portable and can be read by any tool that supports Parquet (Spark, pandas, Polars, BigQuery, Snowflake).

  • Save pipeline.py.

You won't see a change yet. The next chunk wires this function into main() so it runs as part of the pipeline.

Write the analytics function

A data warehouse is only as useful as the questions you can answer with it. This function runs four queries that demonstrate the power of your Gold layer and Parquet exports.

  • In pipeline.py, add the following function below your export_to_parquet(con) function:
def run_analytics(con):
    """Run sample analytical queries against Gold tables."""
    print("\n--- Top 5 Most Active Regions ---")
    results = con.execute("""
        SELECT region, earthquake_count, avg_magnitude, max_magnitude
        FROM gold.regional_activity
        LIMIT 5
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- Magnitude Distribution ---")
    results = con.execute("""
        SELECT magnitude_category, earthquake_count
        FROM gold.magnitude_distribution
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- 5 Most Active Days ---")
    results = con.execute("""
        SELECT event_date, earthquake_count, max_magnitude
        FROM gold.daily_earthquake_summary
        ORDER BY earthquake_count DESC
        LIMIT 5
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- Querying Parquet directly (no import!) ---")
    results = con.execute("""
        SELECT COUNT(*) AS total_earthquakes,
               ROUND(AVG(mag), 2) AS avg_mag
        FROM read_parquet('silver/earthquakes.parquet')
    """).fetchdf()
    print(results.to_string(index=False))

What does this code do?

  • The first three queries hit your Gold tables directly. They surface the top 5 most active regions, the magnitude distribution across categories, and the 5 days with the most seismic activity.
  • The fourth query uses read_parquet('silver/earthquakes.parquet') to query the exported Parquet file without importing it back into a table. DuckDB treats Parquet files as virtual tables.
  • The .fetchdf() method returns results as a pandas DataFrame, giving you nicely formatted table output in the terminal.

Why is querying Parquet directly useful?

In production, data teams often store Parquet files on cloud storage (S3, GCS, Azure Blob) and query them without loading into a database first. This pattern is called a data lakehouse. DuckDB's ability to read Parquet directly means you can run SQL against files anywhere.

  • Save pipeline.py.

You won't see output yet. The next substep wires both new functions into your pipeline's main() so you can run everything end to end.

Update main() and run the full pipeline

Your two new functions exist but aren't called yet. You need to wire them into the main() function so the pipeline runs all five stages in sequence.

  • In pipeline.py, scroll to your main() function.
  • Find these lines:
    print("\n[4/5] Building Gold layer (aggregations)...")
    build_gold(con)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")
  • Replace those lines with the following, which adds the export and analytics stages before closing the connection:
    print("\n[4/5] Building Gold layer (aggregations)...")
    build_gold(con)

    # Export all layers to Parquet files
    print("\n[5/5] Exporting to Parquet...")
    export_to_parquet(con)

    # Run analytical queries to demonstrate the warehouse
    print("\n" + "=" * 50)
    print("ANALYTICAL QUERIES")
    print("=" * 50)
    run_analytics(con)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")

What changed?

  • export_to_parquet(con) writes all five tables to Parquet files in their respective directories.
  • run_analytics(con) queries the Gold tables and demonstrates direct Parquet reads.
  • Both calls happen before con.close() so the database connection is still active.
  • Save pipeline.py.
  • Run the full pipeline by executing this command in your terminal:
python pipeline.py
  • You should see output for all five pipeline stages, followed by formatted tables showing your analytical results.
  • Confirm the Parquet files were created by running:
ls bronze/ silver/ gold/
  • You should see earthquakes_raw.parquet in bronze/, earthquakes.parquet in silver/, and three Parquet files in gold/.

Seeing a ModuleNotFoundError for pandas?

The .fetchdf() method requires pandas. If you see this error, install pandas by running pip install pandas in your terminal, then re-run the pipeline.

help me fix pandas import errors in my DuckDB pipeline

Seeing a FileNotFoundError for the Parquet paths?

Make sure your main() function still has the os.makedirs(folder, exist_ok=True) loop at the top. This creates the bronze/, silver/, and gold/ directories before the export runs.

help me debug file path errors in my Parquet export

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and that all five Parquet files exist in their directories.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file at this point:

import duckdb
import requests
from datetime import datetime, timezone
import os


def extract_earthquake_data():
    """Extract earthquake data from USGS GeoJSON feed (M2.5+ past 30 days)."""
    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    print(f"Extracted {data['metadata']['count']} earthquakes from USGS")
    return data["features"]


def load_bronze(con, features):
    """Load raw earthquake data into the Bronze layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    con.execute("DROP TABLE IF EXISTS bronze.earthquakes_raw")
    con.execute("""
        CREATE TABLE bronze.earthquakes_raw (
            id VARCHAR,
            mag DOUBLE,
            place VARCHAR,
            time_epoch BIGINT,
            updated_epoch BIGINT,
            tz INTEGER,
            url VARCHAR,
            detail VARCHAR,
            felt INTEGER,
            cdi DOUBLE,
            mmi DOUBLE,
            alert VARCHAR,
            status VARCHAR,
            tsunami INTEGER,
            sig INTEGER,
            net VARCHAR,
            code VARCHAR,
            ids VARCHAR,
            sources VARCHAR,
            types VARCHAR,
            nst INTEGER,
            dmin DOUBLE,
            rms DOUBLE,
            gap DOUBLE,
            mag_type VARCHAR,
            event_type VARCHAR,
            longitude DOUBLE,
            latitude DOUBLE,
            depth DOUBLE,
            _ingested_at TIMESTAMP,
            _source_url VARCHAR
        )
    """)

    ingested_at = datetime.now(timezone.utc).isoformat()
    source_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_month.geojson"

    for feature in features:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        con.execute("""
            INSERT INTO bronze.earthquakes_raw VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, [
            feature["id"],
            props.get("mag"),
            props.get("place"),
            props.get("time"),
            props.get("updated"),
            props.get("tz"),
            props.get("url"),
            props.get("detail"),
            props.get("felt"),
            props.get("cdi"),
            props.get("mmi"),
            props.get("alert"),
            props.get("status"),
            props.get("tsunami"),
            props.get("sig"),
            props.get("net"),
            props.get("code"),
            props.get("ids"),
            props.get("sources"),
            props.get("types"),
            props.get("nst"),
            props.get("dmin"),
            props.get("rms"),
            props.get("gap"),
            props.get("magType"),
            props.get("type"),
            coords[0],
            coords[1],
            coords[2],
            ingested_at,
            source_url,
        ])

    count = con.execute("SELECT COUNT(*) FROM bronze.earthquakes_raw").fetchone()[0]
    print(f"Bronze layer: {count} rows loaded")


def transform_silver(con):
    """Transform Bronze data into cleaned Silver layer."""
    con.execute("CREATE SCHEMA IF NOT EXISTS silver")
    con.execute("DROP TABLE IF EXISTS silver.earthquakes")
    con.execute("""
        CREATE TABLE silver.earthquakes AS
        WITH deduplicated AS (
            SELECT *,
                ROW_NUMBER() OVER (
                    PARTITION BY id
                    ORDER BY updated_epoch DESC
                ) AS row_num
            FROM bronze.earthquakes_raw
            WHERE mag IS NOT NULL
        )
        SELECT
            id,
            mag,
            place,
            epoch_ms(time_epoch) AS event_time,
            epoch_ms(updated_epoch) AS updated_at,
            url,
            alert,
            status,
            tsunami,
            sig,
            net,
            mag_type,
            event_type,
            longitude,
            latitude,
            depth,
            felt,
            cdi,
            mmi,
            nst,
            dmin,
            rms,
            gap,
            _ingested_at,
            _source_url
        FROM deduplicated
        WHERE row_num = 1
    """)

    count = con.execute("SELECT COUNT(*) FROM silver.earthquakes").fetchone()[0]
    nulls = con.execute("""
        SELECT COUNT(*) FROM silver.earthquakes
        WHERE event_time IS NULL OR mag IS NULL OR id IS NULL
    """).fetchone()[0]
    print(f"Silver layer: {count} rows (removed {nulls} rows with NULL key fields)")


def build_gold(con):
    """Build Gold layer with business-ready aggregations."""
    con.execute("CREATE SCHEMA IF NOT EXISTS gold")

    # Gold table 1: Daily earthquake summary
    con.execute("DROP TABLE IF EXISTS gold.daily_earthquake_summary")
    con.execute("""
        CREATE TABLE gold.daily_earthquake_summary AS
        SELECT
            CAST(event_time AS DATE) AS event_date,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude,
            ROUND(AVG(depth), 2) AS avg_depth_km
        FROM silver.earthquakes
        GROUP BY CAST(event_time AS DATE)
        ORDER BY event_date DESC
    """)

    # Gold table 2: Magnitude distribution
    con.execute("DROP TABLE IF EXISTS gold.magnitude_distribution")
    con.execute("""
        CREATE TABLE gold.magnitude_distribution AS
        SELECT
            CASE
                WHEN mag < 3.0 THEN 'Minor (2.5-2.9)'
                WHEN mag < 4.0 THEN 'Light (3.0-3.9)'
                WHEN mag < 5.0 THEN 'Moderate (4.0-4.9)'
                WHEN mag < 6.0 THEN 'Strong (5.0-5.9)'
                WHEN mag < 7.0 THEN 'Major (6.0-6.9)'
                ELSE 'Great (7.0+)'
            END AS magnitude_category,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(depth), 2) AS avg_depth_km,
            ROUND(AVG(sig), 0) AS avg_significance
        FROM silver.earthquakes
        GROUP BY magnitude_category
        ORDER BY MIN(mag)
    """)

    # Gold table 3: Regional activity
    con.execute("DROP TABLE IF EXISTS gold.regional_activity")
    con.execute("""
        CREATE TABLE gold.regional_activity AS
        SELECT
            CASE
                WHEN place LIKE '% of %' THEN SPLIT_PART(place, ' of ', 2)
                ELSE place
            END AS region,
            COUNT(*) AS earthquake_count,
            ROUND(AVG(mag), 2) AS avg_magnitude,
            MAX(mag) AS max_magnitude
        FROM silver.earthquakes
        WHERE place IS NOT NULL
        GROUP BY region
        HAVING COUNT(*) >= 3
        ORDER BY earthquake_count DESC
    """)

    for table in ['daily_earthquake_summary', 'magnitude_distribution', 'regional_activity']:
        count = con.execute(f"SELECT COUNT(*) FROM gold.{table}").fetchone()[0]
        print(f"Gold layer: gold.{table} - {count} rows")


def export_to_parquet(con):
    """Export each layer to Parquet files."""
    con.execute("COPY bronze.earthquakes_raw TO 'bronze/earthquakes_raw.parquet' (FORMAT parquet)")
    con.execute("COPY silver.earthquakes TO 'silver/earthquakes.parquet' (FORMAT parquet)")
    con.execute("COPY gold.daily_earthquake_summary TO 'gold/daily_earthquake_summary.parquet' (FORMAT parquet)")
    con.execute("COPY gold.magnitude_distribution TO 'gold/magnitude_distribution.parquet' (FORMAT parquet)")
    con.execute("COPY gold.regional_activity TO 'gold/regional_activity.parquet' (FORMAT parquet)")
    print("Parquet files exported to bronze/, silver/, gold/ directories")


def run_analytics(con):
    """Run sample analytical queries against Gold tables."""
    print("\n--- Top 5 Most Active Regions ---")
    results = con.execute("""
        SELECT region, earthquake_count, avg_magnitude, max_magnitude
        FROM gold.regional_activity
        LIMIT 5
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- Magnitude Distribution ---")
    results = con.execute("""
        SELECT magnitude_category, earthquake_count
        FROM gold.magnitude_distribution
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- 5 Most Active Days ---")
    results = con.execute("""
        SELECT event_date, earthquake_count, max_magnitude
        FROM gold.daily_earthquake_summary
        ORDER BY earthquake_count DESC
        LIMIT 5
    """).fetchdf()
    print(results.to_string(index=False))

    print("\n--- Querying Parquet directly (no import!) ---")
    results = con.execute("""
        SELECT COUNT(*) AS total_earthquakes,
               ROUND(AVG(mag), 2) AS avg_mag
        FROM read_parquet('silver/earthquakes.parquet')
    """).fetchdf()
    print(results.to_string(index=False))


def main():
    """Run the full ETL pipeline."""
    for folder in ['bronze', 'silver', 'gold']:
        os.makedirs(folder, exist_ok=True)

    con = duckdb.connect("earthquake_warehouse.duckdb")

    print("=" * 50)
    print("EARTHQUAKE DATA WAREHOUSE - ETL PIPELINE")
    print("=" * 50)

    print("\n[1/5] Extracting data from USGS API...")
    features = extract_earthquake_data()

    print("\n[2/5] Loading Bronze layer (raw data)...")
    load_bronze(con, features)

    print("\n[3/5] Transforming Silver layer (clean + deduplicate)...")
    transform_silver(con)

    print("\n[4/5] Building Gold layer (aggregations)...")
    build_gold(con)

    print("\n[5/5] Exporting to Parquet...")
    export_to_parquet(con)

    print("\n" + "=" * 50)
    print("ANALYTICAL QUERIES")
    print("=" * 50)
    run_analytics(con)

    con.close()
    print("\n✓ Pipeline complete! Database saved to earthquake_warehouse.duckdb")


if __name__ == "__main__":
    main()

Your earthquake data warehouse is complete. You've taken raw API data through all three Medallion Architecture layers and exported the results as portable Parquet files. Next up, the Secret Mission takes your warehouse to the cloud.

Secret mission

Query Parquet on S3 from DuckDB

Your data warehouse lives on your machine right now. In production, data engineers store Parquet files in cloud object storage and query them remotely. Push your Gold-layer Parquet files to AWS S3 and query them directly from DuckDB using the httpfs extension.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using DuckDB, so there are no ongoing costs for the main pipeline. If you completed the Secret Mission, your S3 bucket may accumulate minor storage charges over time.

Cost warning

If you completed the Secret Mission and created an S3 bucket, the Parquet files stored there will incur small storage charges. The amount is negligible for a few files, but it is best to delete the bucket if you are done experimenting.

Resources you used:

  • Local project folder on your Desktop (contains pipeline.py, s3_query.py, earthquake_warehouse.duckdb, and Parquet directories).
  • Python virtual environment (inside the project folder).
  • S3 bucket with Gold Parquet files (Secret Mission only).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the pipeline further.

  • Your local files remain on your Desktop.
  • The DuckDB database file persists between runs, so you can re-query without re-running the pipeline.
  • If you completed the Secret Mission, your S3 bucket and Parquet files remain in AWS.

Pause - I'll come back to this later

Free up cloud resources while keeping your local files intact so you can pick up later.

  • Your local project folder, database, and Parquet files stay on disk at no cost.
  • If you completed the Secret Mission, delete your S3 bucket to avoid storage charges. You can re-upload the Parquet files later from your local gold/ folder.
  • Delete your S3 bucket and its contents by running:
aws s3 rb s3://[[BUCKET_NAME="your-earthquake-data-bucket"]] --force
  • Deactivate your Python virtual environment by running:
deactivate

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your database, Parquet files, Python scripts, and S3 bucket permanently.

  • If you completed the Secret Mission, delete your S3 bucket and its contents by running:
aws s3 rb s3://[[BUCKET_NAME="your-earthquake-data-bucket"]] --force
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf earthquake-warehouse

On Windows, use this command instead:

cd ~\Desktop
rmdir /s /q earthquake-warehouse
  • Confirm the folder is gone by running:
ls

Nice Work!

Nice Work!

You just built a real data warehouse from scratch. Your pipeline.py pulls live earthquake data from the USGS API, processes it through three architectural layers, and exports analytics-ready Parquet files.

You've learned how to:

  • Built a complete ETL pipeline that extracts live data from the USGS Earthquake API and processes it through the Medallion Architecture (Bronze, Silver, Gold).
  • Created and queried Parquet files, the columnar format that dominates modern data engineering.
  • Wrote SQL transformations in DuckDB that clean, deduplicate, and aggregate raw data into business-ready insights.
  • Extended your warehouse to the cloud by querying S3 Parquet files directly from DuckDB using the httpfs extension.

Ready to quiz yourself?