Build an ETL Pipeline with DuckDB

Build a modular ETL pipeline that loads live crypto data into DuckDB.

Introduction

30 Second Summary

Every dashboard, report, and analytics tool you have ever used is powered by a pipeline that pulls raw data from somewhere, cleans it up, and loads it into a database. When that pipeline breaks or duplicates data, everything downstream falls apart.

In this project, you will build a production-grade ETL pipeline in Python that extracts live cryptocurrency market data, validates it against a data contract, transforms it with derived metrics, and loads it into a DuckDB analytical database.

What You'll Build

Run a single command and watch your pipeline pull 50 live cryptocurrency prices from the CoinGecko API, validate each record, compute market insights like volatility bands and market cap tiers, and land clean data in a queryable database you can explore with SQL.

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

  • A modular pipeline you can run with a single command to extract, validate, transform, and load 50 cryptocurrency records in seconds.
  • An idempotent loading pattern that prevents duplicate records. Run the pipeline ten times and still see exactly 50 rows in the database.
  • A data contract and quarantine system that catches bad records before they corrupt your data and logs exactly why each one failed.
  • Secret Mission: Add incremental loading with watermark tracking so the pipeline only processes records updated since the last run.

Are there any prerequisites?

You should have basic Python experience (writing functions, running scripts from the terminal) and VS Code installed. You will install all project-specific packages (duckdb, requests, pandas) during the first step.

Before We Start

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

Set Up Your Project Environment

A production-grade pipeline needs more than just code that works. It needs a clear folder structure, isolated dependencies, and centralized configuration.

In this step, you will create a modular project layout that mirrors how professional data engineers organize ETL pipelines. Each pipeline stage gets its own file, dependencies are locked to exact versions, and settings live in one place.

In this step, get ready to:
  • Create a project folder with separate modules for each ETL stage.
  • Install and pin your Python dependencies in a virtual environment.
  • Build a centralized configuration module using a Python dataclass.
Create the project folder structure

Every file in your pipeline will live inside a Python package called pipeline. This package has a separate module for each ETL stage: extract, validate, transform, and load.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Open the integrated terminal by pressing Ctrl+` (backtick key, same on both macOS and Windows).
  • Move to your Desktop and create the project folder by running these commands:
cd ~/Desktop
mkdir crypto-etl-pipeline
cd crypto-etl-pipeline
  • Create the pipeline package folder by running:
mkdir pipeline
  • Create the empty module files inside your project by running:

macOS

touch pipeline/__init__.py pipeline/config.py pipeline/extract.py pipeline/validate.py pipeline/transform.py pipeline/load.py main.py

Windows

New-Item -ItemType File -Path pipeline/__init__.py, pipeline/config.py, pipeline/extract.py, pipeline/validate.py, pipeline/transform.py, pipeline/load.py, main.py
  • Confirm the files exist by running:
ls pipeline/

You should see __init__.py, config.py, extract.py, validate.py, transform.py, and load.py listed in the output.

Why this folder structure?

The pipeline/ folder is a Python package (that is what __init__.py signals). Each stage of your ETL pipeline lives in its own module. This means you can test, debug, and run each stage independently.

This modular layout is exactly what hiring managers look for when reviewing data engineering portfolios. It shows you think in composable, maintainable pieces.

  • Open this project folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the crypto-etl-pipeline folder, and click Open.

You should see the pipeline/ folder and main.py listed in the VS Code Explorer sidebar on the left.

Don't see your files in the sidebar?

Make sure you opened the crypto-etl-pipeline folder itself, not its parent (Desktop). The Explorer sidebar should show pipeline/ and main.py at the top level.

If you see a crypto-etl-pipeline folder listed inside the sidebar, you opened the Desktop instead. Close the folder and try File > Open Folder again, this time double-clicking into crypto-etl-pipeline before clicking Open.

help me open my project folder in VS Code correctly

Create a virtual environment and install dependencies

A virtual environment isolates your project's packages from the rest of your system. This guarantees anyone who clones your repo gets the exact same versions you used.

  • Open a new terminal inside VS Code by pressing Ctrl+` (if the terminal is not already open).
  • Create a virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

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

Seeing a permissions error on Windows?

Windows sometimes blocks script execution by default. Run this command first, then try activating again:

Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned

help me activate my Python virtual environment on Windows

  • Install the three project dependencies by running:
pip install duckdb==1.5.5 requests==2.34.2 pandas==2.3.3

What are these packages?

  • duckdb is an embedded analytical database. It runs inside your Python process with no server to manage.
  • requests handles HTTP calls to the CoinGecko API.
  • pandas provides DataFrames for transforming and enriching your data.

You will see pip download and install several packages (including sub-dependencies). The install is complete when the terminal returns to your prompt.

  • Create a requirements.txt file that pins these exact versions. In VS Code's Explorer sidebar, right-click in the empty space below your files and select New File.
  • Name it requirements.txt and press Enter.
  • Paste the following content into the file:
duckdb==1.5.5
requests==2.34.2
pandas==2.3.3

Why pin exact versions?

Pinning versions with == ensures anyone who runs pip install -r requirements.txt gets the identical packages you used. Without version pins, a future install could pull a breaking update and your pipeline silently fails.

  • Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).

Before you verify the install, what do you think you will see when you import duckdb and print its version?

  • Verify your packages installed correctly by running:
python -c "import duckdb; print(duckdb.__version__)"

You should see 1.5.5 printed to the terminal. This confirms DuckDB is installed at the correct version.

Seeing ModuleNotFoundError?

This means your virtual environment is not active. Check that (venv) appears at the start of your terminal prompt. If it does not, re-run the activation command from the platform tab above.

If you are on Windows and see an error about the Microsoft Visual C++ Redistributable, download and install it from the Microsoft website, then run the pip install command again.

help me fix my DuckDB import error

Create the pipeline configuration module

Every setting your pipeline needs (the API URL, how many coins to fetch, the database file path) belongs in one central place. A Python dataclass is perfect for this because it is immutable, self-documenting, and easy to import from any module.

  • In VS Code's Explorer sidebar, expand the pipeline folder and click on config.py to open it.
  • Paste the following code into config.py:
from dataclasses import dataclass


@dataclass(frozen=True)
class PipelineConfig:
    api_base_url: str = "https://api.coingecko.com/api/v3"
    coins_per_page: int = 50
    vs_currency: str = "usd"
    database_path: str = "crypto_market.duckdb"
    request_timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 2.0

What does this code do?

  • @dataclass(frozen=True) makes the config immutable. Once created, no code can accidentally change settings mid-run.
  • Each field has a sensible default. You can override any value by passing it when you create a PipelineConfig() instance.
  • api_base_url points to CoinGecko's free, keyless public API. No signup or API key required.
  • max_retries and retry_delay control how the extractor recovers from rate limits and network errors.
  • Save config.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Verify the config module works by running this in your terminal:
python -c "from pipeline.config import PipelineConfig; print(PipelineConfig())"

You should see the full dataclass printed with all seven fields and their default values.

Seeing an ImportError?

Make sure your terminal's working directory is crypto-etl-pipeline (the folder that contains the pipeline/ directory). Run pwd to check. If you are in the wrong folder, run cd ~/Desktop/crypto-etl-pipeline to get back.

Also confirm that pipeline/__init__.py exists. Python needs this file to recognize the folder as a package.

help me fix my pipeline config import error

✔️ Awesome, I've got everything!

Great. Double check you have saved both config.py and requirements.txt.

ⓧ I'd like to double check the full code

Here is the complete pipeline/config.py:

from dataclasses import dataclass


@dataclass(frozen=True)
class PipelineConfig:
    api_base_url: str = "https://api.coingecko.com/api/v3"
    coins_per_page: int = 50
    vs_currency: str = "usd"
    database_path: str = "crypto_market.duckdb"
    request_timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 2.0

And your requirements.txt:

duckdb==1.5.5
requests==2.34.2
pandas==2.3.3

Your project skeleton is ready. You have a clean modular layout, pinned dependencies, and a central config that every module will import. Next up, you will write the extraction module that pulls live cryptocurrency data from the CoinGecko API.

Extract Live Market Data from CoinGecko

Your project structure is in place and your dependencies are installed. Now you need actual data flowing through the pipeline.

In this step, you will build an extraction module that pulls live cryptocurrency prices from the CoinGecko API. Because the API has rate limits and network calls can fail, you will handle the same concerns every production pipeline faces: timeouts, retries, and exponential backoff.

In this step, get ready to:
  • Build an extraction function that calls the CoinGecko /coins/markets endpoint with retry logic.
  • Add logging that records extraction progress, retries, and record counts.
  • Run the extractor standalone and confirm 50 live cryptocurrency records arrive.
Build the extraction function

The extraction module is responsible for one thing: calling the CoinGecko API and returning raw JSON data. It needs to handle three failure modes that every networked pipeline encounters: timeouts, rate limiting (HTTP 429), and general request failures.

  • In VS Code, open pipeline/extract.py (currently empty) and paste the following code to set up the module and define the function signature:
import logging
import time

import requests

from pipeline.config import PipelineConfig

logger = logging.getLogger(__name__)


def extract_market_data(config: PipelineConfig) -> list[dict]:
    url = f"{config.api_base_url}/coins/markets"
    params = {
        "vs_currency": config.vs_currency,
        "order": "market_cap_desc",
        "per_page": config.coins_per_page,
        "page": 1,
    }

What does this code do?

  • logging and time are standard library modules for recording progress and adding delays between retries.
  • requests is the HTTP client library that handles the actual API call.
  • PipelineConfig is the dataclass you built in Step 1. It provides the API URL, timeout, and retry settings.
  • The params dictionary tells CoinGecko to return the top 50 coins ranked by market cap, priced in USD.
  • Directly below the params dictionary (still inside the function), paste the retry loop that handles rate limiting, timeouts, and failures:
    for attempt in range(1, config.max_retries + 1):
        try:
            logger.info(f"Extracting market data (attempt {attempt}/{config.max_retries})")
            response = requests.get(url, params=params, timeout=config.request_timeout)

            if response.status_code == 429:
                wait_time = config.retry_delay * (2 ** (attempt - 1))
                logger.warning(f"Rate limited (429). Waiting {wait_time}s before retry.")
                time.sleep(wait_time)
                continue

            response.raise_for_status()
            data = response.json()
            logger.info(f"Successfully extracted {len(data)} records")
            return data

        except requests.exceptions.Timeout:
            logger.warning(f"Request timed out (attempt {attempt}/{config.max_retries})")
            if attempt == config.max_retries:
                raise
            time.sleep(config.retry_delay)

        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")
            if attempt == config.max_retries:
                raise
            time.sleep(config.retry_delay)

    return []

What does the retry loop do?

This loop tries the API call up to 3 times before giving up. Each failure mode gets its own handling:

  • HTTP 429 (rate limited): waits with exponential backoff (2s, 4s, 8s) then retries.
  • Timeout: logs a warning and retries. After the final attempt, re-raises the exception so the caller knows extraction failed.
  • Other request errors: same retry-then-raise pattern.
  • return [] at the end is a safety net. If all retries are exhausted without raising, the function returns an empty list.
  • Save pipeline/extract.py.
  • Confirm the module loads without syntax errors by running this command in your terminal:
python -c "from pipeline.extract import extract_market_data; print('Module loads OK')"

You should see Module loads OK printed.

Seeing a syntax error?

  • Check that the for loop is indented one level inside the function, and that return [] is at the same indentation level as the for statement.
  • Make sure you have the closing parenthesis on the params dictionary (the } on its own line).
  • Verify your virtual environment is activated. You should see (venv) in your terminal prompt.

Help me debug a syntax error in my extract.py module

Add the standalone runner

To test the extraction in isolation, you need a runner block that configures logging and prints the results. Python's if __name__ == "__main__" pattern lets you run a module directly without affecting imports from other files.

  • At the very bottom of pipeline/extract.py (after the function, with two blank lines above), paste the standalone runner:
if __name__ == "__main__":
    import json

    logging.basicConfig(level=logging.INFO)
    config = PipelineConfig()
    data = extract_market_data(config)
    print(json.dumps(data[:3], indent=2))
    print(f"\nTotal records extracted: {len(data)}")

What does the runner do?

  • logging.basicConfig(level=logging.INFO) turns on log output so you can see extraction progress and any retry messages in the terminal.
  • json.dumps(data[:3], indent=2) pretty-prints the first 3 records so you can inspect the data shape without flooding the terminal.
  • The final print confirms the total record count matches the 50 you requested.
  • Save pipeline/extract.py.
Run the extractor and confirm live data

Before you run this, what do you expect to see? The function calls the CoinGecko API requesting 50 coins. Think about whether all 50 will arrive or if something might go wrong.

  • Run the extraction module by executing this command:
python -m pipeline.extract

You should see log output followed by JSON data for the first 3 cryptocurrencies, then a count showing 50 total records extracted. The JSON includes fields like id, current_price, market_cap, and last_updated.

What should I see?

Your terminal output looks something like this (prices will differ since it is live data):

[
  {
    "id": "bitcoin",
    "symbol": "btc",
    "name": "Bitcoin",
    "current_price": 97000.0,
    "market_cap": 1920000000000,
    ...
  }
]

Followed by Total records extracted: 50 at the bottom.

Seeing a connection error or timeout?

  • Check your internet connection. The CoinGecko API requires outbound HTTPS access.
  • If you see a 429 status in the logs, the retry logic is working. Wait 30 seconds and run the command again.
  • If you see ModuleNotFoundError: No module named 'requests', make sure your virtual environment is activated before running the command.

Help me troubleshoot my CoinGecko API extraction

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline/extract.py and the output shows 50 records.

ⓧ I'd like to double check the full code

Your pipeline/extract.py is made up of three sections you built above. Scroll up and confirm each is present, in order:

  • Imports, logger, and the extract_market_data function signature with url and params setup.
  • The for attempt retry loop with 429 handling, timeout handling, and general exception handling, ending with return [].
  • The if __name__ == "__main__" standalone runner at the bottom.

Your extraction module is pulling live market data and handling failures gracefully. Next up, you will build a data contract that validates every record before it moves downstream.

Validate Against a Data Contract

Your extractor pulls 50 records from the CoinGecko API, but raw API data is unreliable. Fields can arrive as null. Numeric values can be negative when they shouldn't be. Timestamps can be malformed strings.

If these bad records slip through to your transformation layer, they'll crash your pipeline or produce garbage output. A data contract acts as a quality gate between extraction and transformation, catching problems early and quarantining failures with clear reasons.

In this step, get ready to:
  • Define a data contract specifying required fields, types, and value constraints.
  • Build a validation module that separates valid records from rejected ones.
  • Run the validator on live extracted data and see quarantined records with error messages.
Define the data contract

A data contract is a dictionary that declares what each field must look like. For every field in a CoinGecko record, you'll specify whether it's required, what type it should be, and whether it has a minimum value.

  • In VS Code, create a new file called validate.py inside the pipeline/ folder.
  • Add the imports, logger, and data contract by pasting the following code:
import logging
from datetime import datetime

from pipeline.config import PipelineConfig

logger = logging.getLogger(__name__)


# Define the expected shape and constraints for each field
DATA_CONTRACT = {
    "id": {"type": str, "required": True},
    "symbol": {"type": str, "required": True},
    "name": {"type": str, "required": True},
    "current_price": {"type": (int, float), "required": True, "min_value": 0},
    "market_cap": {"type": (int, float), "required": True, "min_value": 0},
    "total_volume": {"type": (int, float), "required": False},
    "market_cap_rank": {"type": (int, float), "required": False},
    "last_updated": {"type": str, "required": True},
}

What does this code do?

Each key in DATA_CONTRACT maps to a set of rules for that field.

  • required: True means the field must not be None. If it is, the record fails.
  • type declares the expected Python type. Numeric fields accept both int and float.
  • min_value: 0 catches negative prices or market caps that would produce nonsense downstream.
  • Save pipeline/validate.py.

Why a dictionary instead of a library like Pydantic?

Libraries like Pydantic are great for complex schemas, but a plain dictionary keeps the validation logic visible and debuggable. You can read exactly what each check does without learning a framework's decorator syntax first.

Build the record validator

Now you need a function that takes a single record and checks it against every rule in the contract. It returns a tuple: a boolean (pass/fail) and a list of error messages explaining what went wrong.

  • In pipeline/validate.py, add the validate_record function below the DATA_CONTRACT dictionary:
def validate_record(record: dict) -> tuple[bool, list[str]]:
    errors = []

    for field, rules in DATA_CONTRACT.items():
        value = record.get(field)

        # Check if required field is missing
        if rules["required"] and value is None:
            errors.append(f"{field} is null (required)")
            continue

        # Skip optional fields that are absent
        if value is None:
            continue

        # Check type matches
        if not isinstance(value, rules["type"]):
            errors.append(f"{field} has wrong type: expected {rules['type']}, got {type(value)}")
            continue

        # Check minimum value constraint
        if "min_value" in rules and isinstance(value, (int, float)):
            if value < rules["min_value"]:
                errors.append(f"{field} is {value}, below minimum {rules['min_value']}")

    # Validate timestamp format separately
    if record.get("last_updated"):
        try:
            datetime.fromisoformat(record["last_updated"].replace("Z", "+00:00"))
        except (ValueError, AttributeError):
            errors.append(f"last_updated is not a valid ISO timestamp: {record.get('last_updated')}")

    return (len(errors) == 0, errors)

What does this code do?

The function loops through every field defined in the contract and applies each rule in sequence.

  • If a required field is None, it logs the error and skips further checks on that field.
  • Type checks use isinstance so that int and float are both accepted for numeric fields.
  • The timestamp check parses the ISO string to confirm it's a real date, not just any string.
  • Save pipeline/validate.py.
  • Test the function with a quick command in your terminal:
python -c "from pipeline.validate import validate_record; print(validate_record({'id': 'bitcoin', 'symbol': 'btc', 'name': 'Bitcoin', 'current_price': 50000, 'market_cap': 1000000000, 'last_updated': '2024-01-01T00:00:00Z'}))"

You should see (True, []) printed, meaning the record passed all checks with zero errors.

  • Now test with a bad record to see the contract reject it:
python -c "from pipeline.validate import validate_record; print(validate_record({'id': 'test', 'symbol': 'tst', 'name': 'Test', 'current_price': -5, 'market_cap': None, 'last_updated': '2024-01-01T00:00:00Z'}))"

You should see (False, ['market_cap is null (required)', 'current_price is -5, below minimum 0']) showing two specific reasons for rejection.

Seeing a ModuleNotFoundError?

  • Make sure you're running the command from your project's root folder (the one containing the pipeline/ directory), not from inside pipeline/ itself.
  • Confirm your virtual environment is still activated. You should see the environment name in your terminal prompt.

My validate_record function isn't returning the expected output. What should I check?

Build the batch validator and runner

A single-record validator is useful, but your pipeline processes 50 records at once. The batch function loops through all records, sorts them into valid and rejected lists, and logs a summary.

  • In pipeline/validate.py, add the validate_batch function below validate_record:
def validate_batch(records: list[dict]) -> tuple[list[dict], list[dict]]:
    valid_records = []
    rejected_records = []

    for record in records:
        is_valid, errors = validate_record(record)
        if is_valid:
            valid_records.append(record)
        else:
            # Keep the original record and its errors for debugging
            rejected_records.append({
                "record_id": record.get("id", "unknown"),
                "errors": errors,
                "raw_record": record,
            })

    logger.info(f"Validation complete: {len(valid_records)} valid, {len(rejected_records)} rejected")

    if rejected_records:
        for rejected in rejected_records:
            logger.warning(f"Rejected '{rejected['record_id']}': {rejected['errors']}")

    return valid_records, rejected_records

What does this code do?

  • validate_batch calls validate_record on every record and sorts them into two lists: passed and failed.
  • Rejected records keep their original data (raw_record) so you can inspect what went wrong without re-fetching from the API.
  • The logger outputs a summary count and warns for each rejected record by name, creating an audit trail.
  • Now add the standalone runner at the bottom of pipeline/validate.py so you can test the full extract-validate chain:
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    from pipeline.extract import extract_market_data

    config = PipelineConfig()
    raw_data = extract_market_data(config)
    valid, rejected = validate_batch(raw_data)
    print(f"\nValid: {len(valid)} | Rejected: {len(rejected)}")
    if rejected:
        for r in rejected[:5]:
            print(f"  - {r['record_id']}: {r['errors']}")

What does the runner do?

The if __name__ == "__main__" block lets you run the file directly for testing. It calls the extractor from Step 2, feeds the raw data into validate_batch, and prints the results.

  • Save pipeline/validate.py.

✔️ Awesome, I've got everything!

Great. Confirm you've saved the file before running the validator.

ⓧ I'd like to double check the full code

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

  • Imports (logging, datetime, PipelineConfig) and logger setup.
  • DATA_CONTRACT dictionary with 8 fields defined.
  • validate_record function that returns a (bool, list[str]) tuple.
  • validate_batch function that returns (valid_records, rejected_records).
  • if __name__ == "__main__" block that extracts, validates, and prints results.
Run the validator on live data

Time to see the full validation module in action against real CoinGecko data. Before you run it, think about what you expect: will all 50 records pass, or will some get rejected?

  • Run the validation module by executing this command:
python -m pipeline.validate

You should see log output showing the extraction, then a validation summary like this:

INFO:pipeline.extract:Extracting market data (attempt 1/3)
INFO:pipeline.extract:Successfully extracted 50 records
INFO:pipeline.validate:Validation complete: 50 valid, 0 rejected

Valid: 50 | Rejected: 0

What if all records are valid?

CoinGecko's data is generally well-formed, so seeing 50 valid and 0 rejected is the normal result. The value of your contract shows up when data drifts. If CoinGecko ever returns a null market cap for a new coin or a malformed timestamp, your pipeline will catch it immediately instead of crashing in the transform step.

Seeing connection errors or timeouts?

  • Check your internet connection. The validator calls the CoinGecko API through the extractor.
  • If you see a 429 rate limit, wait 30 seconds and try again. CoinGecko limits keyless requests to about 10-30 calls per minute.
  • Make sure pipeline/extract.py is still working by running python -m pipeline.extract first.

My validation module throws an error when I run it. What should I check?

Your pipeline now has a quality gate that catches bad data before it reaches transformations. Next up, you'll build the transformation layer that turns validated records into analytically useful data with derived metrics like market cap tiers and volatility scores.

Transform and Enrich the Data

Your pipeline can now extract live data and validate it against a contract. But validated data still contains raw numbers that are hard to analyze on their own.

In this step, you will build a transformation layer that turns raw price and market cap values into business-ready insights. You will calculate derived metrics like market cap tiers and volatility percentages using composable pure functions.

In this step, get ready to:
  • Build pure transformation functions that classify coins by market cap tier and calculate price volatility.
  • Compose the individual transformations into a single pipeline function.
  • Run the full extract-validate-transform chain and inspect the enriched DataFrame.
Build the pure transformation functions

Each transformation function follows the same pattern: it takes a pandas DataFrame in and returns a new DataFrame out. This makes each function independently testable and reorderable.

  • In VS Code, create a new file called transform.py inside your pipeline/ folder.
  • Add the imports and the first transformation function by pasting the following code:
import logging
from datetime import datetime, timezone

import pandas as pd

logger = logging.getLogger(__name__)


def classify_market_cap_tier(df: pd.DataFrame) -> pd.DataFrame:
    def tier(market_cap):
        if market_cap >= 200_000_000_000:
            return "Mega"
        elif market_cap >= 10_000_000_000:
            return "Large"
        elif market_cap >= 1_000_000_000:
            return "Mid"
        elif market_cap >= 100_000_000:
            return "Small"
        else:
            return "Micro"

    result = df.copy()
    result["market_cap_tier"] = result["market_cap"].apply(tier)
    return result

What does this code do?

The imports bring in pandas for DataFrame operations and datetime for timestamps (used later).

The classify_market_cap_tier function copies the input DataFrame to avoid mutating it, then applies a tier label based on market cap value. Coins worth over $200 billion get "Mega", down to "Micro" for anything under $100 million.

  • Below classify_market_cap_tier, add the volatility calculator and metadata stamper:

def calculate_volatility(df: pd.DataFrame) -> pd.DataFrame:
    result = df.copy()
    result["price_volatility_pct"] = (
        (result["high_24h"] - result["low_24h"]) / result["low_24h"] * 100
    ).round(2)
    result["price_volatility_pct"] = result["price_volatility_pct"].fillna(0.0)
    return result


def add_metadata(df: pd.DataFrame) -> pd.DataFrame:
    result = df.copy()
    result["extracted_at"] = datetime.now(timezone.utc).isoformat()
    return result

What does this code do?

The calculate_volatility function computes a percentage spread between the 24-hour high and low prices. A coin whose high is 10% above its low has 10% volatility. The fillna(0.0) handles coins with missing high/low data gracefully.

The add_metadata function stamps each row with the current UTC time. This tells you exactly when the pipeline processed the data.

  • Save pipeline/transform.py.
  • Verify the functions are syntactically correct by running this import check:
python -c "from pipeline.transform import classify_market_cap_tier, calculate_volatility, add_metadata; print('Functions imported successfully')"

You should see Functions imported successfully printed in your terminal.

Seeing an ImportError or SyntaxError?

  • Make sure you saved transform.py inside the pipeline/ folder, not the project root.
  • Check that pipeline/__init__.py exists. Python needs it to recognize the folder as a package.
  • Confirm your virtual environment is activated. pandas must be installed for the import to succeed.

Help me debug my transform.py import error

Compose the transformations into a single pipeline

Individual functions are useful, but the rest of your pipeline needs a single entry point that runs all transformations in sequence. The transform function will serve as that orchestrator.

  • Below add_metadata, add the composed transform function:

def transform(records: list[dict]) -> pd.DataFrame:
    logger.info(f"Transforming {len(records)} records")

    df = pd.DataFrame(records)

    columns_to_keep = [
        "id", "symbol", "name", "current_price", "market_cap",
        "market_cap_rank", "total_volume", "high_24h", "low_24h",
        "price_change_percentage_24h", "circulating_supply",
        "total_supply", "last_updated",
    ]
    df = df[[col for col in columns_to_keep if col in df.columns]]

    df = classify_market_cap_tier(df)
    df = calculate_volatility(df)
    df = add_metadata(df)

    logger.info(f"Transformation complete: {len(df)} records, {len(df.columns)} columns")
    return df

What does this code do?

The transform function takes raw record dictionaries (the validated output from the previous step) and converts them to a DataFrame. It first prunes the columns down to only the ones the pipeline cares about.

Then it runs the three transformations in sequence: tier classification, volatility calculation, and metadata stamping. Each function receives the DataFrame from the previous one, forming a composable chain.

  • Save pipeline/transform.py.
Run the full extract-validate-transform chain

Now you will add a runnable __main__ block so you can execute the module directly and see the enriched data.

  • At the bottom of pipeline/transform.py, add the following runner block:

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    from pipeline.extract import extract_market_data
    from pipeline.validate import validate_batch

    config = PipelineConfig()
    raw_data = extract_market_data(config)
    valid, _ = validate_batch(raw_data)
    df = transform(valid)
    print(df[["name", "current_price", "market_cap_tier", "price_volatility_pct"]].head(10))

What does this code do?

This runner block wires together the full chain so far: extract from the API, validate against the contract, then transform. It prints a focused subset of columns for the top 10 coins so you can inspect the new market_cap_tier and price_volatility_pct columns.

Missing PipelineConfig import

The __main__ block uses PipelineConfig which is defined in pipeline/config.py. If you see a NameError when running, add from pipeline.config import PipelineConfig right after the other imports inside the if __name__ block.

  • Save pipeline/transform.py.

Before you run this, what do you expect to see in the output? Think about which coins would land in the "Mega" tier and which would have high volatility.

  • Run the transform module by executing this command:
python -m pipeline.transform

You should see a table with 10 rows showing coin names, current prices, market cap tiers, and volatility percentages. Bitcoin and Ethereum should appear as "Mega" tier coins.

Seeing a KeyError or empty DataFrame?

  • If you see KeyError: 'high_24h', the API may have returned records missing that field. The fillna(0.0) in calculate_volatility handles null values, but the column must exist. Confirm columns_to_keep includes high_24h and low_24h.
  • If you see NameError: name 'PipelineConfig' is not defined, add from pipeline.config import PipelineConfig inside the if __name__ block after the other imports.
  • If the table looks empty, check that validate_batch is returning valid records. Re-run python -m pipeline.validate to confirm extraction and validation still work.

Help me debug my transform module output

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline/transform.py and that your terminal output shows all four new columns.

ⓧ I'd like to double check the full code

Here is the complete pipeline/transform.py file. Compare it line by line with yours:

Your file should contain these constructs in order:

  • Imports (logging, datetime, pandas) and logger setup.
  • The classify_market_cap_tier function with the inner tier helper and five threshold levels.
  • The calculate_volatility function computing the high/low spread percentage.
  • The add_metadata function adding a UTC timestamp.
  • The composed transform function that selects columns and chains all three transformations.
  • The if __name__ == "__main__" block that runs the full extract-validate-transform chain and prints 10 rows.

Your transformation layer is complete. Each function is independently testable and the composed pipeline turns raw API data into enriched, analysis-ready records. Next up, you will load this transformed data into a DuckDB database with idempotent upserts that prevent duplicates.

Load into DuckDB with Idempotent Upserts

Your pipeline can now extract, validate, and transform live market data. The final piece is persisting that data somewhere you can query it later.

There is a subtle problem with loading data into a database from a scheduled pipeline. If you run the pipeline twice with the same data, a naive INSERT doubles your row count. In production, this means corrupted analytics, inflated metrics, and silent data quality rot. The fix is idempotency: a load operation that produces the same result whether you run it once or a hundred times.

In this step, get ready to:
  • Build the loader module with a staging table upsert pattern that prevents duplicates.
  • Wire the complete pipeline in main.py with logging at each stage.
  • Run the pipeline twice and verify the row count stays at 50.
Define the database schema

The loader needs to know what columns and types the target table should have. You will define this as a SQL constant at the top of the module.

  • In VS Code, open the empty pipeline/load.py file.
  • Add the imports, logger, and table schema by pasting this code:
import logging

import duckdb
import pandas as pd

from pipeline.config import PipelineConfig

logger = logging.getLogger(__name__)


CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS market_data (
    coin_id VARCHAR,
    symbol VARCHAR,
    name VARCHAR,
    current_price DOUBLE,
    market_cap DOUBLE,
    market_cap_rank INTEGER,
    total_volume DOUBLE,
    high_24h DOUBLE,
    low_24h DOUBLE,
    price_change_percentage_24h DOUBLE,
    circulating_supply DOUBLE,
    total_supply DOUBLE,
    last_updated TIMESTAMP,
    market_cap_tier VARCHAR,
    price_volatility_pct DOUBLE,
    extracted_at TIMESTAMP,
    PRIMARY KEY (coin_id, extracted_at)
)
"""

What does this code do?

The CREATE_TABLE_SQL constant defines the schema for the market_data table. The PRIMARY KEY (coin_id, extracted_at) is a composite key that uniquely identifies each record.

This composite key is what makes deduplication possible. If the same coin is extracted at the same timestamp, the pipeline knows it is a duplicate.

Build the upsert function

The core of idempotent loading is the staging table pattern. Instead of inserting directly into the target table, you load into a temporary staging table first, delete any matching rows from the target, then insert the staged rows. This guarantees that re-running the pipeline with the same data never produces duplicates.

  • Add the load_data function below the CREATE_TABLE_SQL constant:
def load_data(df: pd.DataFrame, config: PipelineConfig) -> int:
    logger.info(f"Loading {len(df)} records into DuckDB")

    con = duckdb.connect(config.database_path)

    con.execute(CREATE_TABLE_SQL)

    df_to_load = df.rename(columns={"id": "coin_id"})

    if "last_updated" in df_to_load.columns:
        df_to_load["last_updated"] = pd.to_datetime(df_to_load["last_updated"])

    if "extracted_at" in df_to_load.columns:
        df_to_load["extracted_at"] = pd.to_datetime(df_to_load["extracted_at"])

    con.execute("CREATE OR REPLACE TEMP TABLE staging AS SELECT * FROM df_to_load")

    con.execute("""
        DELETE FROM market_data
        WHERE coin_id IN (SELECT coin_id FROM staging)
        AND extracted_at IN (SELECT extracted_at FROM staging)
    """)

    result = con.execute("INSERT INTO market_data SELECT * FROM staging").fetchone()
    rows_loaded = con.execute("SELECT count(*) FROM market_data").fetchone()[0]

    logger.info(f"Load complete. Total rows in market_data: {rows_loaded}")

    con.close()
    return rows_loaded

How does the staging table pattern work?

The upsert follows three steps inside one connection:

  • Stage: CREATE OR REPLACE TEMP TABLE staging loads the DataFrame into a temporary table that only lives for this session.
  • Delete matches: the DELETE FROM market_data removes any rows that share the same coin_id and extracted_at as the staged rows.
  • Insert: the fresh staged rows go in cleanly. If the data is new, nothing was deleted. If it is a re-run, the old duplicates were removed first.

DuckDB can reference a pandas DataFrame by variable name directly in SQL thanks to replacement scans. That is why SELECT * FROM df_to_load works without registering the DataFrame manually.

Why a staging table instead of INSERT OR REPLACE?

DuckDB does support INSERT OR REPLACE for simple upserts. The staging pattern gives you more control: you can add validation between staging and inserting, inspect what will be overwritten, or apply more complex merge logic later without restructuring your loader.

  • Add the standalone test runner at the bottom of pipeline/load.py:
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    from pipeline.config import PipelineConfig
    from pipeline.extract import extract_market_data
    from pipeline.validate import validate_batch
    from pipeline.transform import transform

    config = PipelineConfig()
    raw_data = extract_market_data(config)
    valid, _ = validate_batch(raw_data)
    df = transform(valid)
    total = load_data(df, config)
    print(f"Total rows in database: {total}")

What does this test runner do?

This __main__ block lets you run pipeline/load.py in isolation. It chains all four stages (extract, validate, transform, load) and prints the final row count so you can confirm the loader works before wiring up main.py.

  • Save pipeline/load.py.
  • Run the loader standalone by running this command:
python -m pipeline.load

You should see logging output showing the extraction, validation, transformation, and then a final line like Total rows in database: 50.

Seeing an import error or missing module?

  • Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
  • Confirm you are running the command from the project root folder (the one containing the pipeline/ directory).
  • Check that pipeline/__init__.py exists. Even though it is empty, Python needs it to treat pipeline/ as a package.

Help me debug my pipeline/load.py import error

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline/load.py and that you saw the Total rows in database: 50 line in your output.

ⓧ I'd like to double check the full code

Scroll up and confirm each chunk is present in order in your pipeline/load.py:

  • Imports, logger, and CREATE_TABLE_SQL constant (the full schema with all 16 columns and the composite primary key).
  • The load_data function (connect, create table, rename columns, convert timestamps, stage, delete, insert, count, close).
  • The if __name__ == "__main__" block that chains all four stages and prints the total.
Wire the complete pipeline in main.py

With all four modules built, it is time to create the single entry point that orchestrates the entire pipeline. This file will call each stage in order, log progress, and print a summary at the end.

  • Create a new file called main.py in your project root folder (next to the pipeline/ directory).
  • Add the imports and logging configuration:
import logging
import time

from pipeline.config import PipelineConfig
from pipeline.extract import extract_market_data
from pipeline.validate import validate_batch
from pipeline.transform import transform
from pipeline.load import load_data

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("pipeline")

What does this logging format do?

The format string produces structured log lines like 2026-08-05 14:32:01 | INFO | pipeline | PIPELINE START. This makes it easy to scan pipeline output and spot warnings or errors at a glance.

  • Add the first half of the run_pipeline function below the logging config. This handles extraction and validation:
def run_pipeline():
    start_time = time.time()
    config = PipelineConfig()

    logger.info("=" * 60)
    logger.info("PIPELINE START")
    logger.info("=" * 60)

    # Extract
    logger.info("[1/4] Extracting data from CoinGecko API...")
    raw_data = extract_market_data(config)

    # Validate
    logger.info("[2/4] Validating records against data contract...")
    valid_records, rejected_records = validate_batch(raw_data)

    if not valid_records:
        logger.error("No valid records to process. Pipeline aborted.")
        return

What does this code do?

The function starts a timer, then calls extract and validate in sequence. If validation rejects every record, the pipeline aborts early rather than loading empty data.

  • Add the rest of the run_pipeline function and the entry point directly below (continuing the same function body):
    # Transform
    logger.info("[3/4] Transforming and enriching data...")
    df = transform(valid_records)

    # Load
    logger.info("[4/4] Loading into DuckDB...")
    total_rows = load_data(df, config)

    # Summary
    elapsed = time.time() - start_time
    logger.info("=" * 60)
    logger.info("PIPELINE COMPLETE")
    logger.info(f"  Records extracted:  {len(raw_data)}")
    logger.info(f"  Records validated:  {len(valid_records)}")
    logger.info(f"  Records rejected:   {len(rejected_records)}")
    logger.info(f"  Records loaded:     {len(df)}")
    logger.info(f"  Total in database:  {total_rows}")
    logger.info(f"  Duration:           {elapsed:.2f}s")
    logger.info("=" * 60)


if __name__ == "__main__":
    run_pipeline()

What does the summary section do?

After all four stages complete, the pipeline logs a final report showing exactly how many records flowed through each stage and how long it took. This is the kind of observability that production pipelines need for monitoring and debugging.

  • Save main.py.
  • Run the full pipeline by running this command:
python main.py

You should see structured log output showing each stage progressing from PIPELINE START through to PIPELINE COMPLETE, with a summary showing 50 records extracted, validated, and loaded.

Pipeline aborts with no valid records?

  • This usually means the CoinGecko API returned an error. Check the log lines above the abort message for a 429 (rate limit) or timeout warning.
  • Wait 60 seconds and try again. The keyless API has a rate limit of 10-30 calls per minute.

Help me debug why my pipeline aborts with no valid records

✔️ Awesome, I've got everything!

Great. Confirm you saved main.py and that the summary shows Total in database: 50.

ⓧ I'd like to double check the full code

Scroll up and confirm each chunk is present in order in your main.py:

  • Imports (logging, time, plus all four pipeline modules) and logging.basicConfig with the structured format.
  • The run_pipeline function: timer start, config, PIPELINE START banner, extract, validate with the early-return guard, transform, load, and the PIPELINE COMPLETE summary.
  • The if __name__ == "__main__" block calling run_pipeline().
Prove idempotency by running twice

The whole point of the staging table pattern is that running the pipeline again with the same data does not create duplicates. Time to prove it.

Before you run the pipeline a second time, think about what you expect to happen. The pipeline will extract the same 50 coins, validate them, transform them, and attempt to load them. Will the database end up with 50 rows or 100?

  • Run the pipeline again by running this command:
python main.py

Check the summary line. You should see Total in database: 50 again, not 100. The DELETE step removed the matching rows before the INSERT replaced them with identical data.

  • Verify the exact row count with a direct query by running this command:
python -c "import duckdb; con = duckdb.connect('crypto_market.duckdb'); print(con.sql('SELECT count(*) FROM market_data').fetchone())"

You should see (50,) printed. Exactly 50 rows, confirming your upsert logic prevents duplicates.

Why does this matter for production?

Scheduled pipelines run unattended. Network glitches, partial failures, and manual re-runs are inevitable. Without idempotency, every retry inflates your data. With the staging-delete-insert pattern, you can safely re-run anytime without worrying about corrupted analytics.

Seeing more than 50 rows?

  • Make sure the DELETE FROM market_data statement in your load_data function references both coin_id and extracted_at from the staging table.
  • Delete the crypto_market.duckdb file and run python main.py twice from scratch to test cleanly.

Help me debug why my pipeline is creating duplicate rows

Your pipeline is fully wired and idempotent. You can run it any number of times without fear of duplicated data. Next up, you will take on the Secret Mission and add incremental loading with watermark tracking.

Secret mission

Add Incremental Loading with Watermark Tracking

Your pipeline re-processes every record on every run. Production pipelines handling millions of rows cannot afford that. Add watermark tracking so the pipeline only processes records updated since the last successful load.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources on disk or delete them entirely.

Resources you used:

  • Project folder on your Desktop (contains all source code, the pipeline/ package, and main.py).
  • Python virtual environment (inside your project folder).
  • crypto_market.duckdb database file (inside your project folder).

Keep everything running

No action needed. Choose this if you want to keep running the pipeline to accumulate historical cryptocurrency data over time.

  • The crypto_market.duckdb file stays on disk at negligible storage cost (a few megabytes at most).
  • The pipeline only runs when you manually execute python main.py. There is no background process or scheduler.
  • Your virtual environment and installed packages remain available for future runs.

Pause - I'll come back to this later

There is nothing to pause. The pipeline only runs when you execute it manually. No background processes, no servers, no scheduled jobs.

  • Close VS Code when you are done.
  • Your project folder, virtual environment, and database file remain on disk until you choose to delete them.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and the database. Nothing will remain on your machine.

  • If your virtual environment is still active in the terminal, deactivate it by running:
deactivate
  • Delete the project folder from your Desktop. On macOS or Linux, run:
cd ~/Desktop
rm -rf crypto-etl-pipeline

On Windows, run:

cd ~\Desktop
Remove-Item -Recurse -Force crypto-etl-pipeline

Folder name may differ

Replace crypto-etl-pipeline with whatever you named your project folder in Step 1. This single deletion removes all source files, the virtual environment, and the crypto_market.duckdb database.

  • Confirm the folder is gone by running ls ~/Desktop (macOS/Linux) or ls ~\Desktop (Windows). Your project folder should no longer appear.

Nice Work!

Nice Work!

Nice work! You just built a production-grade ETL pipeline that pulls live cryptocurrency data from CoinGecko, validates it against a data contract, transforms it with derived metrics, and loads it into DuckDB.

You've learned how to:

  • Build a modular ETL pipeline with separate extract, validate, transform, and load stages that pulls live data from a free API and loads it into an analytical database you can query with SQL.
  • Implement idempotent upsert logic using a staging table pattern that prevents duplicate records when the pipeline runs multiple times.
  • Design composable transformation functions that calculate derived metrics like market cap tiers and price volatility bands from raw API data.
  • Add incremental loading with watermark tracking so the pipeline only processes records updated since the last successful run (Secret Mission).

Ready to quiz yourself?