Build a USGS Earthquake ETL Pipeline

Transform USGS earthquake data into Parquet and query it in BigQuery.

Introduction

30 Second Summary

Earthquake records keep changing, so a saved spreadsheet can become stale quickly. A column that looks like time can also be stored as plain text, which makes trustworthy analysis harder.

In this project, you will build a repeatable ETL pipeline that pulls current earthquake records from the USGS Earthquake Hazards Program. You will turn its CSV feed into validated Parquet data before loading it into BigQuery for analysis.

What You'll Build

You will demo a fresh earthquake dataset by running etl.py before answering questions about event dates, magnitude groups, or the strongest events in BigQuery.

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

  • A repeatable data refresh that preserves the source CSV while producing a validated Parquet file.
  • Visible schema checks that expose text timestamps before confirming UTC dates, unique IDs, valid coordinates, and matching row counts.
  • SQL analysis that proves date coverage, summarizes magnitude buckets, and surfaces the ten strongest recent events from a native BigQuery table.
  • Secret Mission: Add a configurable minimum-magnitude rule that filters the curated data before safely replacing the BigQuery table.

Are there any prerequisites?

Basic Python syntax is enough for the code in this project. You also need a Google Account for BigQuery Sandbox. The guide covers the Windows workspace plus every remaining dependency.

Before We Start

This opening checkpoint gives your local work plus cloud work a clear purpose before any setup begins. The pipeline preserves the rolling USGS 30-day earthquake feed as raw CSV before loading validated typed Parquet into BigQuery for quality checks plus analytical summaries.

Set Up the Workspace and BigQuery Sandbox

Your earthquake ETL pipeline depends on local tools producing the same result each time. Isolating dependencies keeps package changes from shifting that result.

In this step, you'll prepare Visual Studio Code for the project. A Python virtual environment keeps the pipeline's packages separate from your other projects.

A no-billing BigQuery Sandbox project gives your finished Parquet file a warehouse destination. Setting it up now prevents account work from interrupting the data work later.

In this step, get ready to:
  • Verify Python 3.11 or later inside your Windows workspace.
  • Create an isolated environment with pinned pandas plus PyArrow dependencies.
  • Create the no-billing BigQuery Sandbox project named earthquake-etl-lab.
Verify Python and create the workspace

The project needs Python 3.11 or later because the pinned pandas release requires that minimum version. Checking first prevents a confusing package installation failure.

  • Press the Windows key to open Windows search.
  • Type Visual Studio Code.
  • Press Enter to open Visual Studio Code.
  • Select File from the menu bar.
  • Select Open Folder....
  • Select Desktop in the folder dialog.
  • Select New Folder.
  • Name the folder earthquake-etl.
  • Select Select Folder to open the workspace.

You should see earthquake-etl at the top of the Explorer sidebar. This confirms that every file created in the editor lands inside the project folder.

  • Select View from the menu bar.
  • Select Terminal.
  • Select the dropdown icon in the terminal panel.
  • Select Command Prompt as the shell.
  • Check the available Python version by running this command:
python --version

What does this check show?

The command asks the Python executable available to this terminal for its version. The result determines whether the pinned pandas release can run in your environment.

✔️ I see version 3.11 or higher

Your Python version meets the project's requirement. Continue below to isolate the project dependencies.

ⓧ I see an older version

The installed Python release is too old for pandas 3.0.5. Update Python before creating the virtual environment.

  • Visit the official Python downloads page for Windows.
  • Download the current stable Python 3 release for Windows.
  • Run the downloaded installer to complete the Python setup.
  • Close Visual Studio Code after the installation finishes.
  • Use Windows search to reopen Visual Studio Code.
  • Select File from the menu bar.
  • Select Open Folder....
  • Open the earthquake-etl folder from your Desktop.
  • Repeat the Python version check from the main instructions.

ⓧ Command not found

Windows cannot find a Python installation from this terminal. Install the current stable release before continuing.

  • Visit the official Python downloads page for Windows.
  • Download the current stable Python 3 release for Windows.
  • Run the downloaded installer to complete the Python setup.
  • Close Visual Studio Code after the installation finishes.
  • Use Windows search to reopen Visual Studio Code.
  • Select File from the menu bar.
  • Select Open Folder....
  • Open the earthquake-etl folder from your Desktop.
  • Repeat the Python version check from the main instructions.
  • Create the isolated virtual environment by running this command:
python -m venv .venv

What does this command do?

The command creates a Python environment inside .venv. Packages installed there stay scoped to this earthquake project.

  • Confirm that .venv appears beneath earthquake-etl in the Explorer sidebar.

The new .venv folder is the first visible proof that your workspace is isolated. Your global Python packages remain separate.

Virtual environment missing?

Check that the terminal is using Command Prompt from the earthquake-etl workspace. Repeat the command after confirming Python reports 3.11 or later.

Help me diagnose why the Python virtual environment was not created in my earthquake-etl folder on Windows.

Pin and install the data dependencies

A pinned dependency file records the exact package releases used by the pipeline. Reinstalling from that file recreates the same local data environment.

  • Select the New File... button in the Explorer sidebar.
  • Name the file requirements.txt.
  • Add the pinned dependencies by copying this file content:
pandas==3.0.5
pyarrow==25.0.1

What does this file control?

The first pin selects pandas 3.0.5 for reading plus transforming tabular data.

The second pin selects PyArrow 25.0.1 for writing the curated Parquet file later in the project.

  • Press Ctrl+S to save requirements.txt.
  • Confirm that requirements.txt appears beneath earthquake-etl in the Explorer sidebar.

File showing the wrong name?

Check that the Explorer shows exactly requirements.txt. Rename the file if Windows or the editor added another extension.

Help me check why my requirements file has the wrong name in Visual Studio Code.

✔️ Awesome, I've got everything!

Great. Double-check that you saved requirements.txt before continuing.

ⓧ I'd like to double check the full code

pandas==3.0.5
pyarrow==25.0.1
  • Activate the virtual environment by running this command:
.venv\Scripts\activate

What does activation change?

Activation points Python package commands at the environment inside .venv. New dependencies now install inside this project.

You should see the virtual environment name at the start of the Command Prompt line. That prompt change confirms the environment is active.

  • Install the pinned dependencies by running this command:
pip install -r requirements.txt

What does this installation use?

The command reads each pin from requirements.txt. It installs those releases into the active virtual environment.

  • Inspect both installed package versions by running these commands:
pip show pandas
pip show pyarrow

What do these checks prove?

Each command prints metadata for one package in the active environment. The displayed releases prove that the installation followed your pins.

You should see pandas 3.0.5 in the first result. You should see PyArrow 25.0.1 in the second result.

Package installation failed?

Confirm that the virtual environment name appears at the start of your Command Prompt line. Check that both pins in requirements.txt match the full-file reference above.

Help me diagnose a pinned package installation problem in my Windows virtual environment.

Enable BigQuery Sandbox

This cloud setup stays inside BigQuery Sandbox, so it does not ask for a credit card or billing account. Your local Parquet file gets a warehouse destination without creating a paid resource.

The Google Cloud Console may display account terms during the first sign-in. Complete that account prompt before creating the project.

  • Open the BigQuery page in your browser.
  • Authenticate with your Google Account.
  • Accept the displayed terms when prompted.
  • Select Create project.
  • Enter earthquake-etl-lab as the project name.
  • Select No organization when that option is available.
  • Select Create.

Project creation can take a short moment while Google Cloud prepares the workspace. Keep the page open until earthquake-etl-lab becomes the active project.

Why use the sandbox?

The sandbox supports the dataset creation plus local-file loading used later in this project. It keeps the warehouse stage available without attaching a billing account.

Before the final check, do you expect the BigQuery page to request billing or display the Sandbox notice?

  • Return to the BigQuery page.
  • Confirm that the active project is earthquake-etl-lab.
  • Confirm that the Sandbox notice is visible on the BigQuery page.

The BigQuery page displays the Sandbox notice instead of a billing requirement. The project selector shows earthquake-etl-lab.

Your earlier checks also show Python 3.11 or later. The installed packages report pandas 3.0.5 plus PyArrow 25.0.1.

Your isolated Windows workspace now has its warehouse destination ready. Next, you'll extract the live USGS feed and expose its timestamp schema problem.

Extract the CSV and Reveal the Schema Problem

Your Windows workspace is ready. Your no-billing warehouse destination is also waiting for data.

A direct CSV-to-Parquet conversion moves values between formats. Its schema still needs proof before warehouse loading.

You will preserve the USGS Earthquake feed as a raw snapshot. You will also build an early Parquet output to inspect how the timestamp is stored.

In this step, get ready to:
  • Define the source paths and required earthquake fields.
  • Download the live feed and preserve a raw CSV snapshot.
  • Write an early Parquet file and inspect the timestamp type.
Build the extraction path

A raw data layer preserves the source before any cleaning changes it. This gives you evidence of exactly what arrived from the feed.

  • In the VS Code Explorer sidebar, select the new-file button for the earthquake-etl folder.
  • Name the file etl.py.
  • Add the source configuration by copying this code into etl.py:
from pathlib import Path

import pandas as pd

SOURCE_URL = (
    "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv"
)
RAW_CSV_PATH = Path("data/raw/earthquakes.csv")
PARQUET_PATH = Path("data/processed/earthquakes.parquet")

REQUIRED_COLUMNS = [
    "id",
    "time",
    "updated",
    "latitude",
    "longitude",
    "depth",
    "mag",
    "place",
    "type",
    "status",
]

What does this code set up?

  • The Path class represents the local raw and processed file locations.
  • The pandas import gives the script tools for reading CSV data.
  • The SOURCE_URL constant points to the rolling 30-day earthquake feed.
  • The REQUIRED_COLUMNS list defines the source fields that the pipeline depends on.
  • Save etl.py.
  • Confirm etl.py appears beside requirements.txt in the Explorer sidebar.

File missing from the Explorer?

Confirm you created etl.py inside the earthquake-etl folder. A file created outside that folder will not appear beside requirements.txt.

Help me locate or create etl.py in my earthquake-etl folder.

The extraction function reads the live feed into a DataFrame. It also checks the source fields before preserving the raw snapshot.

  • In etl.py, place your cursor after the closing bracket of REQUIRED_COLUMNS.
  • Add the extraction function by copying this code below the list:
def extract() -> pd.DataFrame:
    raw = pd.read_csv(SOURCE_URL)
    missing_columns = set(REQUIRED_COLUMNS) - set(raw.columns)
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"USGS feed is missing required columns: {missing}")

    RAW_CSV_PATH.parent.mkdir(parents=True, exist_ok=True)
    raw.to_csv(RAW_CSV_PATH, index=False)
    print(f"Extracted {len(raw):,} rows to {RAW_CSV_PATH}")
    print(f"Raw time dtype: {raw['time'].dtype}")
    return raw

What does the extraction function do?

  • The pd.read_csv() call downloads the feed into a DataFrame.
  • The missing-column check stops the pipeline if a required source field is absent.
  • The parent-folder call creates data/raw before the script writes the snapshot.
  • The two printed lines expose the downloaded row count and the inferred time dtype.
  • Save etl.py.
  • Confirm the extract() function ends with return raw at the indentation shown above.
Produce the early Parquet result

The first Parquet write carries the extracted frame straight into columnar storage. Reading the file back lets the pipeline inspect the stored result.

  • In etl.py, place your cursor beneath return raw.
  • Add the Parquet writer and run sequence by copying this code below extract():
def write_naive_parquet(raw: pd.DataFrame) -> None:
    PARQUET_PATH.parent.mkdir(parents=True, exist_ok=True)
    raw.to_parquet(
        PARQUET_PATH,
        engine="pyarrow",
        compression="snappy",
        index=False,
    )
    reloaded = pd.read_parquet(PARQUET_PATH, engine="pyarrow")
    print(f"Wrote {len(reloaded):,} rows to {PARQUET_PATH}")
    print(f"Naive time dtype: {reloaded['time'].dtype}")


def main() -> None:
    raw = extract()
    write_naive_parquet(raw)


if __name__ == "__main__":
    main()

What does this code do?

  • The write_naive_parquet() function creates the processed-data folder.
  • The Parquet writer uses PyArrow with Snappy compression.
  • The read-back step confirms that the generated file can be loaded again.
  • The main() function runs extraction before writing the early Parquet result.
  • Save etl.py.
Inspect the timestamp type

A value can look like a timestamp while still carrying a text-based dtype. The run below reveals what the direct conversion actually preserved.

Before you run the pipeline, do you think timestamp-looking values automatically become datetimes during this direct conversion?

  • Run the complete extraction and Parquet path in the activated terminal with this command:
python etl.py

What should you see?

Your terminal reports a downloaded row count and a written Parquet row count. These counts should match because this early conversion has not removed any records.

The Raw time dtype and Naive time dtype lines show a non-datetime type. That result exposes the schema problem on the main path.

  • Expand the data/raw folder in the Explorer sidebar.
  • Select earthquakes.csv.
  • Locate the time column in the raw snapshot.
  • Return to the terminal from earlier.
  • Compare the timestamp-looking values with the printed non-datetime dtype.
  • Confirm earthquakes.parquet appears inside data/processed.

You have your first live pipeline result. It now preserves the USGS feed in both raw and processed local layers.

You also caught the timestamp typing gap before warehouse loading. The next transformation can fix a problem you have already seen with your own data.

Pipeline did not create both files?

Confirm the terminal still shows the activated .venv environment. The installed pandas and PyArrow packages belong to that environment.

Check that your internet connection can reach the USGS feed. The extraction stage needs the live CSV response before it can create either output file.

Help me debug my earthquake ETL extraction or Parquet write.

✔️ Awesome, I've got everything!

Your raw CSV and early Parquet file are ready. Double-check that you saved etl.py.

ⓧ I'd like to double check the full code

from pathlib import Path

import pandas as pd

SOURCE_URL = (
    "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv"
)
RAW_CSV_PATH = Path("data/raw/earthquakes.csv")
PARQUET_PATH = Path("data/processed/earthquakes.parquet")

REQUIRED_COLUMNS = [
    "id",
    "time",
    "updated",
    "latitude",
    "longitude",
    "depth",
    "mag",
    "place",
    "type",
    "status",
]


def extract() -> pd.DataFrame:
    raw = pd.read_csv(SOURCE_URL)
    missing_columns = set(REQUIRED_COLUMNS) - set(raw.columns)
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"USGS feed is missing required columns: {missing}")

    RAW_CSV_PATH.parent.mkdir(parents=True, exist_ok=True)
    raw.to_csv(RAW_CSV_PATH, index=False)
    print(f"Extracted {len(raw):,} rows to {RAW_CSV_PATH}")
    print(f"Raw time dtype: {raw['time'].dtype}")
    return raw


def write_naive_parquet(raw: pd.DataFrame) -> None:
    PARQUET_PATH.parent.mkdir(parents=True, exist_ok=True)
    raw.to_parquet(
        PARQUET_PATH,
        engine="pyarrow",
        compression="snappy",
        index=False,
    )
    reloaded = pd.read_parquet(PARQUET_PATH, engine="pyarrow")
    print(f"Wrote {len(reloaded):,} rows to {PARQUET_PATH}")
    print(f"Naive time dtype: {reloaded['time'].dtype}")


def main() -> None:
    raw = extract()
    write_naive_parquet(raw)


if __name__ == "__main__":
    main()

Your extraction path works and its schema weakness is visible. Next up, you will transform the source into a validated warehouse-ready dataset.

Transform and Validate the Data

Your first Parquet file proved that the extraction path works. It also exposed the source timestamps as text.

The transform stage gives your ETL pipeline a curated schema. Quality checks then stop incomplete or geographically invalid records from reaching the warehouse.

In this step, get ready to:
  • Parse source values into warehouse-ready types.
  • Reject incomplete or invalid earthquake records.
  • Rebuild the curated Parquet file with a verified round trip.
Enforce the curated schema

Pandas can convert the timestamp fields into UTC-aware values. It can also coerce malformed numeric values into missing values that the cleanup stage can remove.

  • In etl.py, find the closing bracket of REQUIRED_COLUMNS.
  • Add the numeric column list plus the magnitude helper below it by pasting this code:
NUMERIC_COLUMNS = ["latitude", "longitude", "depth", "mag"]


def magnitude_bucket(magnitude: float) -> str:
    if magnitude < 3.0:
        return "under_3"
    if magnitude < 5.0:
        return "3_to_under_5"
    if magnitude < 7.0:
        return "5_to_under_7"
    return "7_plus"

What does this code do?

  • The NUMERIC_COLUMNS list keeps the fields that need numeric conversion in one place.
  • The magnitude_bucket() helper assigns each valid magnitude to one neutral reporting bucket.
  • The ordered checks ensure each magnitude receives exactly one label.
  • Save etl.py.
  • Check that the new helper loads without a syntax error by running this command:
python etl.py

What should you see?

You should see the downloaded row count. You should also see Naive time dtype: with no traceback.

Python loads every function definition before it calls main(). A successful run confirms that the new helper is syntactically valid.

Seeing a syntax error?

Check that NUMERIC_COLUMNS sits after the closing bracket of REQUIRED_COLUMNS. Make sure each return line remains indented inside magnitude_bucket().

Help me fix the magnitude helper in etl.py.

The helper defines the reporting categories. The next function applies the complete type conversion and cleanup sequence.

  • In etl.py, find the end of extract().
  • Add transform() below extract() by pasting this code:
def transform(raw: pd.DataFrame) -> pd.DataFrame:
    cleaned = raw[REQUIRED_COLUMNS].copy()
    cleaned["time"] = pd.to_datetime(
        cleaned["time"], format="ISO8601", errors="coerce", utc=True
    )
    cleaned["updated"] = pd.to_datetime(
        cleaned["updated"], format="ISO8601", errors="coerce", utc=True
    )

    for column in NUMERIC_COLUMNS:
        cleaned[column] = pd.to_numeric(cleaned[column], errors="coerce")

    cleaned = cleaned.dropna(
        subset=["id", "time", "updated", "latitude", "longitude", "depth", "mag"]
    )
    cleaned = cleaned.drop_duplicates(subset=["id"], keep="last")
    cleaned["magnitude_bucket"] = cleaned["mag"].map(magnitude_bucket)
    return cleaned.sort_values(by="time", ascending=False)

How does the transformation work?

  • The first line copies only the fields listed in REQUIRED_COLUMNS.
  • The two pd.to_datetime() calls parse ISO 8601 values as UTC timestamps.
  • The numeric loop converts invalid source values into missing values through errors="coerce".
  • The cleanup keeps complete records plus the latest row for each event ID.
  • The final lines add magnitude_bucket before sorting the newest earthquakes first.
  • Save etl.py.
  • Confirm the script still loads every function successfully by running this command:
python etl.py

What does this check prove?

You should see the extraction output plus the existing naive Parquet output. You should not see a traceback from transform().

The old main() still controls this run. The final substep connects the new transformation to the output path.

Does the script stop while loading?

Check that both pd.to_datetime() calls sit inside transform(). Confirm that the for loop uses NUMERIC_COLUMNS exactly.

Help me debug the transform function in etl.py.

Add fail-fast quality checks

Type conversion can still leave a dataset that is empty or geographically impossible. Fail-fast checks turn those conditions into clear errors before the Parquet file is replaced.

  • In etl.py, find the end of transform().
  • Add validate() below transform() by pasting this code:
def validate(cleaned: pd.DataFrame) -> None:
    if cleaned.empty:
        raise ValueError("Validation failed: no rows remain after transformation")
    if len(cleaned["id"]) != len(set(cleaned["id"])):
        raise ValueError("Validation failed: duplicate event IDs remain")
    if not all(-90 <= value <= 90 for value in cleaned["latitude"]):
        raise ValueError("Validation failed: latitude is outside -90 to 90")
    if not all(-180 <= value <= 180 for value in cleaned["longitude"]):
        raise ValueError("Validation failed: longitude is outside -180 to 180")
    print("Validation passed")

What do these checks protect?

  • The empty-frame check stops the pipeline when transformation removes every record.
  • The ID check ensures that each earthquake appears once in the curated output.
  • The latitude check enforces the range from -90 to 90.
  • The longitude check enforces the range from -180 to 180.
  • The success message gives the completed pipeline a visible quality checkpoint.
  • Save etl.py.
  • Check that the validation function loads without a syntax error by running this command:
python etl.py

What should happen now?

The existing extraction path should complete without a traceback. This confirms that Python can load the new validation function.

The final wiring calls validate() with the curated frame. That run produces the visible Validation passed checkpoint.

Seeing an indentation error?

Make sure each raise ValueError line is nested beneath its matching if condition. Keep print("Validation passed") aligned with the four conditions.

Help me fix the validation function in etl.py.

Rebuild and inspect the Parquet file

A round trip writes the curated frame through PyArrow before reading it back. Matching row counts prove that the saved artifact preserves every validated record.

  • Replace the entire write_naive_parquet() function with this final writer:
def write_parquet(cleaned: pd.DataFrame) -> None:
    PARQUET_PATH.parent.mkdir(parents=True, exist_ok=True)
    cleaned.to_parquet(
        PARQUET_PATH,
        engine="pyarrow",
        compression="snappy",
        index=False,
    )
    reloaded = pd.read_parquet(PARQUET_PATH, engine="pyarrow")
    if len(reloaded) != len(cleaned):
        raise ValueError("Parquet round-trip row count does not match")

    print(f"Wrote {len(reloaded):,} rows to {PARQUET_PATH}")
    print(f"Raw CSV size: {RAW_CSV_PATH.stat().st_size:,} bytes")
    print(f"Parquet size: {PARQUET_PATH.stat().st_size:,} bytes")
    print(f"Curated time dtype: {reloaded['time'].dtype}")
    print(reloaded.dtypes)
    print(reloaded[["time", "place", "mag", "depth", "magnitude_bucket"]].head(5))

What does the final writer verify?

  • The writer saves the curated rows with the pyarrow engine.
  • Snappy compression keeps the Parquet artifact compact.
  • The reload checks that the saved row count matches the curated frame.
  • The final output reports both file sizes plus the curated timestamp type.
  • The five-row sample makes the transformed values visible in the terminal.
  • In etl.py, find the existing main() function.
  • Replace the entire function with this final pipeline sequence:
def main() -> None:
    raw = extract()
    cleaned = transform(raw)
    validate(cleaned)
    write_parquet(cleaned)

How does the pipeline flow?

The raw frame preserves the downloaded source snapshot. The cleaned frame holds only transformed records.

Validation runs before write_parquet(). Invalid curated data therefore cannot replace the processed artifact.

  • Save etl.py.

Before you run the finished pipeline, what dtype do you expect for time after the Parquet round trip?

  • Run the completed ETL pipeline with this command:
python etl.py

What should you see?

  • You should see the current USGS row count after extraction.
  • You should see Validation passed before the Parquet write summary.
  • You should see matching curated and reloaded row counts in the pipeline output.
  • You should see a UTC-aware dtype beside Curated time dtype:.
  • You should see the final column types plus five transformed earthquake records.

Pipeline did not reach validation?

Confirm that main() calls transform(raw) before validate(cleaned). Check that the old write_naive_parquet() function has been fully replaced.

Help me debug the completed earthquake ETL pipeline.

That schema gap is closed. Your processed Parquet file now carries validated earthquake records with UTC timestamps.

✔️ Awesome, I've got everything!

Your transformed pipeline is saved. Keep the latest terminal output available for the checkpoint below.

ⓧ I'd like to double check the full code

from pathlib import Path

import pandas as pd

SOURCE_URL = (
    "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv"
)
RAW_CSV_PATH = Path("data/raw/earthquakes.csv")
PARQUET_PATH = Path("data/processed/earthquakes.parquet")

REQUIRED_COLUMNS = [
    "id",
    "time",
    "updated",
    "latitude",
    "longitude",
    "depth",
    "mag",
    "place",
    "type",
    "status",
]
NUMERIC_COLUMNS = ["latitude", "longitude", "depth", "mag"]


def magnitude_bucket(magnitude: float) -> str:
    if magnitude < 3.0:
        return "under_3"
    if magnitude < 5.0:
        return "3_to_under_5"
    if magnitude < 7.0:
        return "5_to_under_7"
    return "7_plus"


def extract() -> pd.DataFrame:
    raw = pd.read_csv(SOURCE_URL)
    missing_columns = set(REQUIRED_COLUMNS) - set(raw.columns)
    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(f"USGS feed is missing required columns: {missing}")

    RAW_CSV_PATH.parent.mkdir(parents=True, exist_ok=True)
    raw.to_csv(RAW_CSV_PATH, index=False)
    print(f"Extracted {len(raw):,} rows to {RAW_CSV_PATH}")
    print(f"Raw time dtype: {raw['time'].dtype}")
    return raw


def transform(raw: pd.DataFrame) -> pd.DataFrame:
    cleaned = raw[REQUIRED_COLUMNS].copy()
    cleaned["time"] = pd.to_datetime(
        cleaned["time"], format="ISO8601", errors="coerce", utc=True
    )
    cleaned["updated"] = pd.to_datetime(
        cleaned["updated"], format="ISO8601", errors="coerce", utc=True
    )

    for column in NUMERIC_COLUMNS:
        cleaned[column] = pd.to_numeric(cleaned[column], errors="coerce")

    cleaned = cleaned.dropna(
        subset=["id", "time", "updated", "latitude", "longitude", "depth", "mag"]
    )
    cleaned = cleaned.drop_duplicates(subset=["id"], keep="last")
    cleaned["magnitude_bucket"] = cleaned["mag"].map(magnitude_bucket)
    return cleaned.sort_values(by="time", ascending=False)


def validate(cleaned: pd.DataFrame) -> None:
    if cleaned.empty:
        raise ValueError("Validation failed: no rows remain after transformation")
    if len(cleaned["id"]) != len(set(cleaned["id"])):
        raise ValueError("Validation failed: duplicate event IDs remain")
    if not all(-90 <= value <= 90 for value in cleaned["latitude"]):
        raise ValueError("Validation failed: latitude is outside -90 to 90")
    if not all(-180 <= value <= 180 for value in cleaned["longitude"]):
        raise ValueError("Validation failed: longitude is outside -180 to 180")
    print("Validation passed")


def write_parquet(cleaned: pd.DataFrame) -> None:
    PARQUET_PATH.parent.mkdir(parents=True, exist_ok=True)
    cleaned.to_parquet(
        PARQUET_PATH,
        engine="pyarrow",
        compression="snappy",
        index=False,
    )
    reloaded = pd.read_parquet(PARQUET_PATH, engine="pyarrow")
    if len(reloaded) != len(cleaned):
        raise ValueError("Parquet round-trip row count does not match")

    print(f"Wrote {len(reloaded):,} rows to {PARQUET_PATH}")
    print(f"Raw CSV size: {RAW_CSV_PATH.stat().st_size:,} bytes")
    print(f"Parquet size: {PARQUET_PATH.stat().st_size:,} bytes")
    print(f"Curated time dtype: {reloaded['time'].dtype}")
    print(reloaded.dtypes)
    print(reloaded[["time", "place", "mag", "depth", "magnitude_bucket"]].head(5))


def main() -> None:
    raw = extract()
    cleaned = transform(raw)
    validate(cleaned)
    write_parquet(cleaned)


if __name__ == "__main__":
    main()

Your curated artifact is ready for warehouse ingestion. Next, you will load its embedded schema into BigQuery.

Load Parquet into BigQuery

Your local ETL run has produced a validated Parquet file with UTC timestamps.

Loading that curated artifact into BigQuery completes the pipeline's load stage. The file's embedded schema lets BigQuery create typed columns without manual definitions.

In this step, get ready to:
  • Create the earthquake_etl dataset in the US multi-region.
  • Upload data/processed/earthquakes.parquet as the native earthquakes table.
  • Confirm the inferred timestamp fields and populated earthquake rows.
Create the destination dataset

A BigQuery dataset groups related tables within one location. The US multi-region keeps this project's warehouse resources together.

  • Switch back to the BigQuery page from earlier.
  • Select the earthquake-etl-lab project in the Explorer panel.
  • Click View actions.
  • Select Create dataset.
  • Enter earthquake_etl in Dataset ID.
  • Select Multi-region under Location type.
  • Choose US (multiple regions in United States) as the location.
  • Leave the remaining settings at their defaults.
  • Select Create dataset.

You'll see earthquake_etl beneath earthquake-etl-lab in the Explorer panel. This confirms that the US multi-region destination is ready.

Dataset missing from Explorer?

  • Refresh the BigQuery page.
  • Expand earthquake-etl-lab in the Explorer panel.
  • Confirm that your selected project is earthquake-etl-lab.

Help me find a newly created BigQuery dataset that is missing from Explorer.

Upload the curated Parquet file

Parquet stores column types inside the file. BigQuery reads that information to create the destination table with an inferred schema.

Why use a local upload?

This workflow uses a local batch upload so each extract, transform, and load stage stays visible.

It avoids adding Cloud Storage or service-account credentials to this first pipeline. The curated Parquet file can move directly from your computer into BigQuery.

  • Select the earthquake_etl dataset in the Explorer panel.
  • Choose Create table.

You'll see the Create table page with separate source and destination settings.

  • Set Create table from to Upload.
  • Click Browse under Select file.
  • Select data/processed/earthquakes.parquet from the earthquake-etl folder.

You'll see the selected Parquet filename beside the file control.

  • Set File format to Parquet.

BigQuery can now read the schema embedded in the curated file.

  • Confirm that Project shows earthquake-etl-lab.
  • Confirm that Dataset shows earthquake_etl.

These destination values place the table inside the dataset you just created.

  • Enter earthquakes in the Table field.
  • Keep Table type set to Native table.

The destination is now set to earthquake_etl.earthquakes as a native BigQuery table.

  • Leave the Schema section unchanged.
  • Select Create table.

Great work. Your curated Parquet artifact is now the native earthquake_etl.earthquakes table.

Table upload not completing?

  • Confirm that data/processed/earthquakes.parquet still exists inside earthquake-etl.
  • Confirm that the local file is no larger than 100 MB.
  • Confirm that File format is set to Parquet.

Help me troubleshoot a BigQuery local Parquet upload.

Inspect the inferred schema and rows

The table's schema proves whether the local transformation survived the warehouse load. Its preview confirms that the curated records are available for analysis.

Before you inspect the table, which local columns do you expect BigQuery to recognize as timestamp fields?

  • Select earthquakes beneath the earthquake_etl dataset.
  • Select the Schema tab.

You'll see time and updated listed as TIMESTAMP fields.

You'll also see id, latitude, longitude, depth, mag, place, type, status, and magnitude_bucket in the inferred schema.

  • Select the Preview tab.

You'll see event IDs, coordinates, depths, magnitudes, places, types, statuses, magnitude buckets, and UTC timestamps populated across the rows.

Your validated Parquet data is now queryable in BigQuery. Next, you'll use SQL to reconcile row counts and produce earthquake summaries.

Prove the Pipeline with SQL

Your curated Parquet file is now a native BigQuery table. The local schema has survived the move into your warehouse.

A table preview confirms that rows exist. SQL checks the complete table for uniqueness, time coverage, magnitude patterns, plus the strongest recent events.

In this step, get ready to:
  • Reconcile the warehouse rows with the latest local ETL output.
  • Summarize earthquake magnitudes through analytical queries.

Why validate again in BigQuery?

Local validation proves that the curated file passed your quality rules. Warehouse queries prove that the load preserved those results.

This second check catches differences between the artifact you produced and the table you plan to analyze.

Run reconciliation and quality queries

A reconciliation query compares the total rows with the distinct earthquake IDs. It also measures the time range covered by the loaded data.

  • Right-click the earthquake-etl folder in Visual Studio Code's file sidebar.
  • Choose the option for creating a file.
  • Name the new file queries.sql.

You should see queries.sql inside the earthquake-etl folder.

  • Add the reconciliation statement to queries.sql by pasting this query:
SELECT
  COUNT(*) AS row_count,
  COUNT(DISTINCT id) AS unique_event_count,
  MIN(time) AS earliest_event_time,
  MAX(time) AS latest_event_time
FROM `earthquake_etl.earthquakes`;

What does this query prove?

  • The row_count value measures every loaded record.
  • The unique_event_count value measures distinct earthquake IDs.
  • The earliest_event_time value marks the beginning of the table's time coverage.
  • The latest_event_time value marks the end of the table's time coverage.
  • Save queries.sql.

Your reconciliation query is now stored locally for repeatable warehouse checks.

  • Return to the BigQuery page from earlier.
  • Start a new BigQuery SQL query tab.

You should see an empty query editor for the earthquake-etl-lab project.

  • Copy the first statement from queries.sql.
  • Paste the statement into the query editor.

Before you execute it, predict whether the two counts will match. Hold that answer in mind.

  • Execute the reconciliation statement.

You should see equal values for row_count plus unique_event_count. You should also see populated values for the earliest event time plus the latest event time.

  • Switch back to the Visual Studio Code terminal from earlier.
  • Compare row_count with the row number in the latest Wrote output from your ETL run.

You have your first warehouse proof. The loaded table preserves one unique ID per earthquake record.

Do the reconciliation counts differ?

Confirm that the query references earthquake_etl.earthquakes. A different table can contain data from another run.

Compare the warehouse count with the same local run that produced the uploaded Parquet file.

Ask for help if the values still differ. Help me reconcile my local ETL row count with my BigQuery table count.

Produce a demonstrable analysis

Magnitude buckets turn individual earthquake records into a compact distribution. The average magnitude gives each bucket an analytical summary.

  • Switch back to queries.sql in Visual Studio Code.
  • Place your cursor two lines below the first query's final semicolon.
  • Add the magnitude summary by pasting this query:
SELECT
  magnitude_bucket,
  COUNT(*) AS event_count,
  ROUND(AVG(mag), 2) AS average_magnitude
FROM `earthquake_etl.earthquakes`
GROUP BY magnitude_bucket
ORDER BY average_magnitude DESC;

What does this aggregation show?

  • The GROUP BY clause creates one result row for each magnitude bucket.
  • The event_count value shows how many earthquakes belong to each bucket.
  • The average_magnitude value summarizes each bucket to two decimal places.
  • The descending order places the bucket with the highest average magnitude first.
  • Save queries.sql.
  • Copy the second statement from queries.sql.

Your local query file now contains the reconciliation check plus the magnitude summary.

  • Return to the existing BigQuery query tab.
  • Paste the second statement below the first statement.

The query tab now holds both saved analytical checks.

  • Select only the second statement in the query tab.
  • Execute the selected aggregation statement.

You should see one row for each populated magnitude_bucket. Each row should include an event count plus an average magnitude.

Missing a magnitude bucket?

A missing bucket can mean the current rolling feed contains no events in that magnitude range.

If the query fails, confirm that magnitude_bucket plus mag appear in the table schema.

Ask for help with unexpected aggregation results. Help me understand my BigQuery magnitude-bucket results.

A bucket summary explains the overall distribution. A top-events query gives you specific earthquakes to show during a demonstration.

  • Switch back to queries.sql in Visual Studio Code.
  • Place your cursor two lines below the second query's final semicolon.
  • Add the strongest-events analysis by pasting this query:
SELECT
  time,
  place,
  mag,
  depth,
  magnitude_bucket
FROM `earthquake_etl.earthquakes`
WHERE mag IS NOT NULL
ORDER BY mag DESC, time DESC
LIMIT 10;

How are the strongest events selected?

  • The WHERE clause excludes rows without a magnitude.
  • The first descending sort places the highest magnitudes first.
  • The second descending sort places the newest event first when magnitudes match.
  • The LIMIT 10 clause keeps the result focused on ten events.
  • Save queries.sql.
  • Copy the third statement from queries.sql.

Your local file now contains all three warehouse checks.

  • Return to the existing BigQuery query tab.
  • Paste the third statement below the second statement.

Before you execute it, predict which event will appear first. Hold that answer in mind.

  • Select only the third statement in the query tab.
  • Execute the selected top-events statement.

You should see ten events ordered from highest magnitude to lowest magnitude. Events with equal magnitudes should appear from newest to oldest.

Your pipeline now has warehouse evidence. You can show its row integrity, time coverage, magnitude distribution, plus strongest recent events.

Do the top events look out of order?

Confirm that the final clause uses ORDER BY mag DESC, time DESC. Reversing either direction changes the result order.

Confirm that you executed only the third statement when checking the strongest events.

Ask for help with an unexpected result order. Help me troubleshoot the ordering in my BigQuery top-events query.

✔️ Awesome, I've got everything!

Great. Double-check that queries.sql is saved with all three statements.

ⓧ I'd like to double check the full code

SELECT
  COUNT(*) AS row_count,
  COUNT(DISTINCT id) AS unique_event_count,
  MIN(time) AS earliest_event_time,
  MAX(time) AS latest_event_time
FROM `earthquake_etl.earthquakes`;

SELECT
  magnitude_bucket,
  COUNT(*) AS event_count,
  ROUND(AVG(mag), 2) AS average_magnitude
FROM `earthquake_etl.earthquakes`
GROUP BY magnitude_bucket
ORDER BY average_magnitude DESC;

SELECT
  time,
  place,
  mag,
  depth,
  magnitude_bucket
FROM `earthquake_etl.earthquakes`
WHERE mag IS NOT NULL
ORDER BY mag DESC, time DESC
LIMIT 10;

Secret mission

Add a Minimum-Magnitude Rule and Rerun the Pipeline

Turn your fixed transformation into a configurable pipeline that keeps earthquakes at or above a chosen magnitude. Rebuild the local outputs, replace the warehouse table, and prove that stale rows are gone.

Clean Up Your Resources

Clean Up Your Resources

This project has no ongoing costs. Choose whether to keep your resources, pause your work, or delete everything.

Resources you used:

  • A local earthquake-etl folder created in Visual Studio Code containing .venv, requirements.txt, etl.py, queries.sql, threshold_check.sql, data/raw/earthquakes.csv, and data/processed/earthquakes.parquet.
  • A Google Cloud project named earthquake-etl-lab.
  • A BigQuery dataset named earthquake_etl containing the native table earthquake_etl.earthquakes.

Keep everything running

No action is needed. Choose this if you want to refresh the earthquake feed or demonstrate the pipeline again.

  • Keep the earthquake-etl folder so the ETL script and SQL queries remain available.
  • Keep earthquake-etl-lab so the filtered table remains queryable.
  • Rerun etl.py whenever you want to refresh the raw CSV and curated Parquet files.
  • Use queries.sql to demonstrate the reconciled row counts and earthquake summaries.

Pause - I'll come back to this later

No worker or scheduled job remains running. Pausing ends your current editor session while preserving the project resources.

  • Close the terminal panel in Visual Studio Code to end the activated .venv session.
  • Close Visual Studio Code when you finish.
  • Leave the earthquake-etl-lab project unchanged.

Your local files remain ready for the next session. The sandbox table expires automatically after 60 days.

Delete - I don't want to use this again

Deleting the local copies is permanent. The steps below shut down the dedicated cloud project and remove the entire local folder.

Deletion Removes the Full Project

Shutting down earthquake-etl-lab includes the earthquake_etl dataset and earthquake_etl.earthquakes table.

Deleting earthquake-etl removes your scripts, virtual environment, raw CSV, and processed Parquet file.

  • Switch back to Google Cloud Console.
  • Select IAM & Admin from the navigation menu.
  • Select Manage resources.
  • Select the checkbox beside earthquake-etl-lab.
  • Select Delete in the toolbar.
  • Enter earthquake-etl-lab in the confirmation field.
  • Select Shut down.

You should see confirmation that the project shutdown was requested. The dataset and table are covered by this project-level cleanup.

  • Switch back to Visual Studio Code.
  • Right-click earthquake-etl in the Explorer sidebar.
  • Select Reveal in File Explorer.
  • Close Visual Studio Code.
  • Right-click the highlighted earthquake-etl folder in File Explorer.
  • Select Delete.
  • Confirm the deletion if Windows asks for approval.
  • Confirm that earthquake-etl no longer appears in File Explorer.

That completes the cleanup. Your local ETL files and dedicated cloud resources are no longer active.

Nice Work!

Nice Work!

Pipeline complete! You built a repeatable earthquake ETL workflow that turns the live USGS CSV feed into validated Parquet data for BigQuery analysis.

You've learned how to:

  • Build a repeatable ETL workflow that preserves a raw CSV snapshot. Convert timestamp text into UTC datetime values before writing a typed Parquet file.
  • Apply data quality checks to reject incomplete records. Catch duplicate IDs plus invalid coordinates before they reach the warehouse. Load the curated artifact into a native BigQuery table using its embedded schema.
  • Use analytical SQL to reconcile local output against warehouse counts. Prove event IDs remain unique. Summarize magnitude buckets plus the strongest recent events.
  • Secret Mission: Added a configurable minimum-magnitude rule at 2.5. Safely replaced the existing BigQuery table. Confirmed events_below_threshold returns zero.

Ready to quiz yourself?