Build an ELT Pipeline with Python

Extract real earthquake data from USGS and load it into PostgreSQL.

Introduction

30 Second Summary

Every day, hundreds of earthquakes shake the planet. Scientists collect this data in real time, but the raw feed is a tangle of nested coordinates, timestamps, and metadata. Turning that mess into something useful is what data engineers do.

In this project, you will build a Python pipeline that pulls live earthquake data from the US Geological Survey, loads it into a local PostgreSQL database, and transforms it into clean tables you can query with simple SQL.

What You'll Build

You run a single command and watch 200+ real earthquakes flow from a government API into your database, then query them like any other table to answer questions about magnitude, location, and depth.

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

  • A working ELT pipeline that extracts live earthquake data from the USGS API, loads raw JSON into PostgreSQL, and transforms it into typed analytical columns.
  • A locally running PostgreSQL database you installed and configured yourself, with a clean earthquakes table you can query to find the biggest quake today or the deepest tremor in the last 24 hours.
  • The hands-on experience of seeing why raw JSONB is painful to query directly, and how a well-designed transform step turns it into something beautiful.
  • Secret Mission: Add incremental loading so your pipeline only fetches new earthquakes on re-runs.

Are there any prerequisites?

You should be comfortable writing basic SQL queries (SELECT, WHERE, ORDER BY) and basic Python (variables, functions, running scripts from the terminal). No prior database administration or data engineering experience is needed.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters. Taking a moment to articulate your goal makes every step that follows more intentional.

Extract Earthquake Data from the USGS API

Your database is ready and waiting for data. Before you can load or transform anything, you need to pull data from a live source.

The first stage of any ELT pipeline is extraction. In this step, you will call the US Geological Survey's real-time earthquake API and see hundreds of live events flow into your terminal.

In this step, get ready to:
  • Create pipeline.py with the extract() function.
  • Run the pipeline and see live earthquake data extracted from the USGS API.
  • Explore the nested JSON structure that your pipeline will need to transform later.
Create pipeline.py with the extract function

The USGS publishes a free GeoJSON feed of every earthquake recorded in the past 24 hours. Your extract() function will call that feed, parse the JSON response, and return the data for the next stage.

  • In Cursor, create a new file called pipeline.py inside your project folder.
  • Add the following imports, configuration constants, and extract function:
import httpx
import psycopg
import json

# Connection string for your local PostgreSQL database
DB_URL = "postgresql://localhost:5432/earthquake_db"
# USGS real-time earthquake feed (past 24 hours)
USGS_FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.geojson"


def extract():
    """Extract earthquake data from USGS GeoJSON feed."""
    print("Extracting earthquake data from USGS...")
    response = httpx.get(USGS_FEED_URL)
    response.raise_for_status()
    data = response.json()
    count = data["metadata"]["count"]
    print(f"Extracted {count} earthquakes")
    return data

What does this code do?

  • DB_URL stores the connection string for your local PostgreSQL database. You will use this in later steps.
  • USGS_FEED_URL points to the USGS GeoJSON feed that is updated every minute with the latest earthquake data worldwide.
  • httpx.get() makes an HTTP GET request to the USGS API and raise_for_status() throws an error if the request fails.
  • data["metadata"]["count"] pulls the total number of earthquakes from the response metadata.
  • Save pipeline.py.
  • Confirm the file has no syntax errors by running:
python -c "import pipeline"

No output means the module loaded without errors.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is still activated. You should see (venv) or similar at the start of your terminal prompt.
  • Double check the import names match exactly: httpx, psycopg, json.
  • help me debug a ModuleNotFoundError when importing pipeline.py

Now add the entry point that will run your pipeline stages in order.

  • At the bottom of pipeline.py, below the extract() function, add the main block:
if __name__ == "__main__":
    print("=" * 50)
    print("ELT Pipeline: USGS Earthquake Data")
    print("=" * 50)

    data = extract()

What does this code do?

The if __name__ == "__main__" guard means this code only runs when you execute pipeline.py directly (not when importing it as a module).

It prints a banner, then calls extract() to fetch the earthquake data. In later steps, you will add load() and transform() calls here to build the full pipeline.

  • Save pipeline.py.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

import httpx
import psycopg
import json

DB_URL = "postgresql://localhost:5432/earthquake_db"
USGS_FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.geojson"


def extract():
    """Extract earthquake data from USGS GeoJSON feed."""
    print("Extracting earthquake data from USGS...")
    response = httpx.get(USGS_FEED_URL)
    response.raise_for_status()
    data = response.json()
    count = data["metadata"]["count"]
    print(f"Extracted {count} earthquakes")
    return data


if __name__ == "__main__":
    print("=" * 50)
    print("ELT Pipeline: USGS Earthquake Data")
    print("=" * 50)

    data = extract()
Run the script and see live earthquake data

Your extract function is ready. Before you run it, think about what you expect to see printed.

  • What do you think the output will look like? Remember that extract() makes a live HTTP request to the USGS API and prints the earthquake count.
  • Run your pipeline:
python pipeline.py

You should see output like this:

==================================================
ELT Pipeline: USGS Earthquake Data
==================================================
Extracting earthquake data from USGS...
Extracted 230 earthquakes

The exact number will vary since the USGS feed updates every minute with real seismic activity.

Seeing a connection error?

  • Check that you have an internet connection. The script calls the USGS API over HTTPS.
  • If you see httpx.ConnectError, try opening the USGS feed URL in your browser to confirm access.
  • help me debug a connection error when calling the USGS earthquake API with httpx
Explore the nested earthquake data

The USGS API returns GeoJSON, a format with deeply nested structure. Understanding this structure now will help you design the transformation step later.

  • Peek at a single earthquake feature by running:
python -c "
import pipeline, json
d = pipeline.extract()
print(json.dumps(d['features'][0], indent=2))
"

You should see a JSON object with three main sections: properties (magnitude, place, time), geometry (longitude, latitude, depth), and id.

What is GeoJSON structure?

Each earthquake "feature" nests data several layers deep. The magnitude is at feature["properties"]["mag"] and the coordinates are at feature["geometry"]["coordinates"] (an array of [longitude, latitude, depth]).

This nested structure is why loading raw JSON into the database first makes sense. Trying to flatten all of this before loading would be fragile and hard to maintain.

You have live earthquake data flowing from the USGS API into your Python script. Next up, you will load this raw data into PostgreSQL so you can query it with SQL.

Load Raw Data into PostgreSQL

In the last step, you extracted live earthquake data from the USGS API and printed it to your terminal. Now you need somewhere to put that data so you can query and transform it later.

The key principle of ELT is to load raw data first, exactly as it arrives from the source. This preserves the original structure so you can re-transform without ever re-extracting. You will store each earthquake as raw JSON in PostgreSQL, then try to query it directly to see why that alone is not enough.

In this step, get ready to:
  • Add a load() function that creates a table and inserts each earthquake as JSONB.
  • Run the pipeline and confirm data landed in PostgreSQL.
  • Query the raw JSONB directly to experience the problem it creates.
Add the load() function

Your pipeline currently extracts data but discards it when the script ends. You need a function that takes the extracted data and writes it into a PostgreSQL table.

  • In Cursor, open pipeline.py and add the following load() function below the extract() function (above the if __name__ block):
def load(data):
    """Load raw earthquake data into PostgreSQL as JSONB."""
    print("Loading raw data into PostgreSQL...")

    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS raw_earthquakes (
                    id TEXT PRIMARY KEY,
                    raw_data JSONB NOT NULL,
                    loaded_at TIMESTAMPTZ DEFAULT NOW()
                )
            """)

            inserted = 0
            for feature in data["features"]:
                cur.execute(
                    """
                    INSERT INTO raw_earthquakes (id, raw_data)
                    VALUES (%s, %s)
                    ON CONFLICT (id) DO NOTHING
                    """,
                    (feature["id"], json.dumps(feature)),
                )
                inserted += cur.rowcount

            conn.commit()

    print(f"Loaded {inserted} new earthquakes into raw_earthquakes")

What does this code do?

This function handles the "L" in ELT.

  • CREATE TABLE IF NOT EXISTS ensures the table exists on the first run but does nothing on re-runs.
  • The table has three columns: id (the USGS earthquake ID), raw_data (the entire GeoJSON feature stored as JSONB), and loaded_at (a timestamp recording when you loaded it).
  • ON CONFLICT (id) DO NOTHING skips duplicates so you can safely re-run the pipeline without errors.
  • psycopg.connect() with a with block auto-commits on success or rolls back on failure.

Now you need to call load() from your main block so the pipeline actually runs it.

  • In pipeline.py, scroll down to the if __name__ == "__main__" block at the bottom of the file.
  • Find this line:
    data = extract()
  • Add load(data) on the next line, so the block now looks like this:
    data = extract()
    load(data)
  • Save pipeline.py.
Run the pipeline and confirm data landed

Your pipeline now extracts from the API and loads into PostgreSQL in one shot. Before you run it, what do you think the output will say?

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

You should see output like this:

==================================================
ELT Pipeline: USGS Earthquake Data
==================================================
Extracting earthquake data from USGS...
Extracted 230 earthquakes
Loading raw data into PostgreSQL...
Loaded 230 new earthquakes into raw_earthquakes

Why does the count vary?

The USGS feed updates every minute with the past 24 hours of seismic data. Your exact count will differ from 230 depending on when you run the script.

Now confirm the data actually made it into PostgreSQL.

  • Connect to your database by running:
psql earthquake_db
  • Check how many rows landed by running this query:
SELECT COUNT(*) FROM raw_earthquakes;

You should see a count matching the number your pipeline reported (around 230).

Seeing 0 rows or a table not found error?

  • Make sure the PostgreSQL service is running. Run brew services start postgresql@17 if it stopped.
  • Confirm you are connected to earthquake_db and not the default postgres database. Check your prompt says earthquake_db=#.
  • Verify that pipeline.py includes the load(data) call in the __main__ block.

Help me debug why my raw_earthquakes table is empty

Query the raw JSONB to see the problem

Your data is safely stored. Now try to answer a simple question: what are the magnitudes of the first 5 earthquakes?

  • Still in psql, run this query:
SELECT id, raw_data->'properties'->>'mag' FROM raw_earthquakes LIMIT 5;

You should see earthquake IDs alongside magnitude values extracted from the JSON. It works, but look at what you had to write to get a single field out.

  • Now try a slightly more complex question. Run this to get magnitudes, places, and depths:
SELECT
    raw_data->'properties'->>'mag' AS magnitude,
    raw_data->'properties'->>'place' AS place,
    raw_data->'geometry'->'coordinates'->>2 AS depth
FROM raw_earthquakes
LIMIT 5;

Every field requires chaining -> and ->> operators. All values come back as text strings, so you cannot sort numerically or filter by magnitude range without casting. There is no way to add an index on any of these fields for efficient queries.

Why is this painful?

The -> operator navigates into a JSON object and returns JSON. The ->> operator does the same but returns plain text. To reach magnitude you need raw_data->'properties'->>'mag' every single time.

This is exactly why the "T" in ELT exists. Raw storage preserves your source data, but you need a transformation step to make it actually usable for analysis.

  • Exit psql by running:
\q

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that running it produces the "Loaded" message.

ⓧ I'd like to double check the full code

Your pipeline.py should now contain these sections in order:

  • Imports (httpx, psycopg, json) and constants (DB_URL, USGS_FEED_URL).
  • The extract() function that fetches data from the USGS API.
  • The load(data) function that creates raw_earthquakes and inserts each feature as JSONB.
  • The if __name__ == "__main__" block calling extract() then load(data).

Scroll up and confirm each section matches what you pasted. If anything looks off, compare your code against the chunks above and fix any differences.

Your raw data is safely in PostgreSQL, but querying it is a chore. Next up, you will write SQL transformations that turn this messy JSON into a clean, typed table where every query is simple and fast.

Transform Raw JSON into Clean Tables

In the last step, you tried querying the raw JSONB data directly. Every field required arrow operators, types came back as strings, and even a basic magnitude lookup felt painful.

That pain is the reason the Transform step exists in an ELT pipeline. You will now write SQL that extracts, casts, and structures the nested JSON into proper typed columns. Once this step is done, every future query will be as simple as querying a normal table.

In this step, get ready to:
  • Add a transform() function that creates a clean earthquakes table with properly typed columns.
  • Run the full pipeline end-to-end (extract, load, transform) and verify the clean table.
  • Run analytical queries on real earthquake data using plain SQL.
Define the clean table schema

The transform function has two jobs: define the target schema (what columns and types you want) and populate it from the raw data. You will start with the schema.

  • In pipeline.py, find the line print(f"Loaded {inserted} new earthquakes into raw_earthquakes") at the end of the load() function.
  • Add the following transform() function below it, leaving a blank line between the two functions:
def transform():
    """Transform raw JSONB into a clean analytical table."""
    print("Transforming data...")

    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS earthquakes (
                    id TEXT PRIMARY KEY,
                    magnitude NUMERIC,
                    place TEXT,
                    occurred_at TIMESTAMPTZ,
                    longitude NUMERIC,
                    latitude NUMERIC,
                    depth_km NUMERIC,
                    earthquake_type TEXT,
                    status TEXT,
                    tsunami_alert BOOLEAN,
                    significance INTEGER,
                    mag_type TEXT,
                    network TEXT,
                    updated_at TIMESTAMPTZ
                )
            """)

What does this code do?

This is the first half of your transform() function. It connects to the database and creates the earthquakes table if it does not already exist.

  • Every column has a proper type: NUMERIC for magnitudes and coordinates, TIMESTAMPTZ for timestamps, BOOLEAN for the tsunami alert flag.
  • Compare this to the raw table where everything was buried inside a single JSONB blob. Now the database engine can sort, filter, and index these values natively.
Add the INSERT...SELECT transformation query

The schema is the destination. Now you need the query that reads from raw_earthquakes and extracts each field into the correct column.

  • Directly below the CREATE TABLE block you just added (still inside the with conn.cursor() as cur: block), add this INSERT...SELECT query:
            cur.execute("""
                INSERT INTO earthquakes (
                    id, magnitude, place, occurred_at,
                    longitude, latitude, depth_km,
                    earthquake_type, status, tsunami_alert,
                    significance, mag_type, network, updated_at
                )
                SELECT
                    id,
                    (raw_data->'properties'->>'mag')::NUMERIC,
                    raw_data->'properties'->>'place',
                    TO_TIMESTAMP(
                        (raw_data->'properties'->>'time')::BIGINT / 1000.0
                    ),
                    (raw_data->'geometry'->'coordinates'->>0)::NUMERIC,
                    (raw_data->'geometry'->'coordinates'->>1)::NUMERIC,
                    (raw_data->'geometry'->'coordinates'->>2)::NUMERIC,
                    raw_data->'properties'->>'type',
                    raw_data->'properties'->>'status',
                    (raw_data->'properties'->>'tsunami')::INTEGER = 1,
                    (raw_data->'properties'->>'sig')::INTEGER,
                    raw_data->'properties'->>'magType',
                    raw_data->'properties'->>'net',
                    TO_TIMESTAMP(
                        (raw_data->'properties'->>'updated')::BIGINT / 1000.0
                    )
                FROM raw_earthquakes

What does this query do?

This is the heart of your transformation. An INSERT...SELECT reads every row from raw_earthquakes and extracts specific fields from the nested JSON into typed columns.

  • The -> operator navigates into a JSON object. The ->> operator extracts a value as text.
  • Casts like ::NUMERIC and ::BIGINT convert the text into proper database types.
  • TO_TIMESTAMP() converts Unix epoch milliseconds into a human-readable TIMESTAMPTZ value.
  • The tsunami field is converted to a BOOLEAN by checking if the integer value equals 1.
Add the ON CONFLICT clause and close the function

If you run the pipeline twice, the same earthquakes would try to insert again. An ON CONFLICT clause tells PostgreSQL to update existing rows instead of failing.

  • Directly below the FROM raw_earthquakes line you just added, paste the following to complete the SQL statement and close the function:
                ON CONFLICT (id) DO UPDATE SET
                    magnitude = EXCLUDED.magnitude,
                    place = EXCLUDED.place,
                    occurred_at = EXCLUDED.occurred_at,
                    longitude = EXCLUDED.longitude,
                    latitude = EXCLUDED.latitude,
                    depth_km = EXCLUDED.depth_km,
                    earthquake_type = EXCLUDED.earthquake_type,
                    status = EXCLUDED.status,
                    tsunami_alert = EXCLUDED.tsunami_alert,
                    significance = EXCLUDED.significance,
                    mag_type = EXCLUDED.mag_type,
                    network = EXCLUDED.network,
                    updated_at = EXCLUDED.updated_at
            """)

            conn.commit()

            cur.execute("SELECT COUNT(*) FROM earthquakes")
            count = cur.fetchone()[0]

    print(f"Transformed data: {count} earthquakes in clean table")

What does ON CONFLICT DO UPDATE do?

When a row with the same id already exists, PostgreSQL updates it with the incoming values instead of throwing an error. The EXCLUDED keyword refers to the row that was proposed for insertion but conflicted.

This makes your pipeline idempotent. You can run it multiple times safely, and the clean table always reflects the latest data.

Update the main block to run the full pipeline

Your transform() function is ready but nothing calls it yet. You need to add it to the pipeline's main block.

  • Scroll down to the if __name__ == "__main__": block at the bottom of pipeline.py.
  • Find this line:
    load(data)
  • Add the following two lines directly below load(data), so the main block now ends like this:
    load(data)
    transform()

    print("\nPipeline complete!")
  • Save pipeline.py.

What does the updated main block do?

The pipeline now runs all three stages in sequence: extract data from the USGS API, load raw JSON into PostgreSQL, then transform it into clean typed columns. This is the full ELT pattern in action.

Seeing a SyntaxError or IndentationError?

Make sure transform() is indented at the same level as load(data) (four spaces). The print("\nPipeline complete!") line should also be at four spaces, with a blank line above it.

help me fix an IndentationError in my pipeline.py main block

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Your pipeline.py should now contain these sections in order. Scroll through your file and confirm each one is present:

  • Imports (httpx, psycopg, json) and constants (DB_URL, USGS_FEED_URL).
  • The extract() function that calls the USGS API and returns the response data.
  • The load(data) function that creates raw_earthquakes and inserts JSONB rows.
  • The transform() function with CREATE TABLE for earthquakes, then the INSERT...SELECT with ON CONFLICT, then commit and count.
  • The if __name__ == "__main__": block calling extract(), load(data), transform(), and printing "Pipeline complete!".
Run the full pipeline end-to-end

Your pipeline now has all three stages wired together. Time to run it and see data flow through the entire extract-load-transform sequence.

  • Before you run this, predict: what output do you expect from each of the three stages?
  • Run the full pipeline by executing this command in your terminal:
python pipeline.py

You should see output like this:

==================================================
ELT Pipeline: USGS Earthquake Data
==================================================
Extracting earthquake data from USGS...
Extracted 230 earthquakes
Loading raw data into PostgreSQL...
Loaded 0 new earthquakes into raw_earthquakes
Transforming data...
Transformed data: 230 earthquakes in clean table

Pipeline complete!

Why does it say 0 new earthquakes loaded?

Your load() function uses ON CONFLICT (id) DO NOTHING, so earthquakes already in raw_earthquakes from your previous run are skipped. The pipeline is idempotent. The transform step still processes all rows into the clean table regardless.

  • Verify the clean table exists by running this in your terminal:
psql earthquake_db -c "SELECT COUNT(*) FROM earthquakes;"

You should see a count matching the number of earthquakes reported in the pipeline output.

Seeing an error about the earthquakes table?

If you see a "relation does not exist" error, the transform() function did not run successfully. Check that transform() is called in the main block after load(data), and re-run python pipeline.py.

help me debug why my earthquakes table was not created after running the pipeline

Run analytical queries on the clean table

The whole point of the Transform step is to make querying effortless. No more arrow operators, no more casting in every query. Try a few analytical queries to see the difference.

  • Find the top 5 earthquakes by magnitude by running:
psql earthquake_db -c "SELECT place, magnitude, depth_km FROM earthquakes ORDER BY magnitude DESC LIMIT 5;"

You should see a clean table with place names, magnitudes, and depths. No JSON operators in sight.

  • Count earthquakes by region by running:
psql earthquake_db -c "SELECT place, COUNT(*) as quake_count FROM earthquakes GROUP BY place ORDER BY quake_count DESC LIMIT 10;"
  • Find the average depth grouped by magnitude range by running:
psql earthquake_db -c "SELECT CASE WHEN magnitude < 1 THEN 'micro' WHEN magnitude < 3 THEN 'minor' WHEN magnitude < 5 THEN 'moderate' ELSE 'strong' END as category, ROUND(AVG(depth_km), 1) as avg_depth, COUNT(*) as count FROM earthquakes GROUP BY category ORDER BY avg_depth DESC;"

Compare this to the raw JSONB queries

Remember the painful JSON extraction from the previous step? That same magnitude filter would have been (raw_data->'properties'->>'mag')::NUMERIC > 4.0. Now it is just magnitude > 4.0.

This is the payoff of the Transform step. You do the messy extraction once, and every query after that is clean SQL.

  • Before you run this final verification, what do you expect to see when filtering for significant earthquakes above magnitude 4.0?
  • Run the step verification query:
psql earthquake_db -c "SELECT place, magnitude, depth_km FROM earthquakes WHERE magnitude > 4.0 ORDER BY magnitude DESC;"

You should see a clean, readable result set of significant earthquakes with proper column names and typed values. If the past day had no earthquakes above 4.0, try lowering the threshold to 2.0.

Query returns 0 rows?

The USGS feed covers only the past 24 hours. On quiet days there may be no earthquakes above magnitude 4.0. Try changing > 4.0 to > 2.0 or > 1.0 to confirm data is present.

help me verify that my earthquakes table has data and understand why my query returns empty results

Your ELT pipeline is complete. You have working extract, load, and transform stages that turn a live API response into a clean, queryable analytical table. Next up, you will push this pipeline further with an optional Secret Mission.

Secret mission

Add Incremental Loading

Your pipeline re-downloads every earthquake on every run. In production, that wastes time and API calls. Can you make it smart enough to fetch only what's new?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • A local PostgreSQL database called earthquake_db (takes minimal disk space, no ongoing cost).
  • The PostgreSQL service running via Homebrew.
  • Your project folder containing pipeline.py and the Python virtual environment.

Keep everything running

No action needed. Choose this if you're still actively building or want to keep querying your earthquake data.

  • PostgreSQL continues running in the background using minimal resources.
  • Your earthquake data persists between runs.
  • Run python pipeline.py anytime to refresh data from the USGS.

Pause - I'll come back to this later

Stop the PostgreSQL service to free up memory. Your database files and project folder stay intact.

  • Stop the PostgreSQL service by running:
brew services stop postgresql@17
  • Restart anytime with:
brew services start postgresql@17

Your earthquake_db database and all project files remain on disk, ready for when you come back.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Drop the earthquake database by running:
dropdb earthquake_db
  • Stop the PostgreSQL service by running:
brew services stop postgresql@17
  • Uninstall PostgreSQL from your machine by running:
brew uninstall postgresql@17
  • Delete your project folder (the one containing pipeline.py) from your Desktop by dragging it to the Trash, or by running rm -rf on the folder in Terminal.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade ELT pipeline that pulls live earthquake data from a government API, stores it as raw JSON, and transforms it into clean analytical tables.

You've learned how to:

  • Installed and configured PostgreSQL on your Mac from scratch and created a database you can connect to from Python.
  • Built a Python ELT pipeline that extracts live earthquake data from the USGS API and loads it as raw JSONB into PostgreSQL.
  • Wrote SQL transformations that convert messy nested JSON into clean, typed analytical tables you can query with standard SQL.
  • Secret Mission: Added incremental loading using the USGS query API and timestamp tracking so your pipeline only fetches new earthquakes on re-runs.

Ready to quiz yourself?