Build an End-to-End Crypto Data Pipeline

Introduction

30 Second Summary

Cryptocurrency markets generate data around the clock. Data engineering teams need reliable, automated pipelines to capture that data, clean it, store it efficiently, and turn it into actionable insights.

In this project, you will build a production-grade data pipeline that ingests cryptocurrency market data from a live API, transforms it through multiple processing layers, and powers a dashboard with real-time market trends.

What You'll Build

A fully orchestrated data pipeline using Airflow, PySpark, MinIO, and DuckDB that follows the Medallion Architecture pattern (Bronze, Silver, and Gold layers) to process crypto data end-to-end.

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

• A fully orchestrated Airflow environment (via the Astro CLI) extracting crypto data from the CoinCap API, transforming it with PySpark, and loading analytical tables into a DuckDB warehouse.
• A Metabase dashboard visualizing crypto market trends (top assets by market cap, 24h volume leaders, and price distributions) directly from your warehouse.
• A GitHub Actions CI/CD workflow that lints DAGs, runs pytest tests, and builds your pipeline image on every push.
• Secret Mission: Add incremental loading with deduplication and a Slowly Changing Dimension (SCD Type 2) table tracking price history changes over time.

Are there any prerequisites?

This project assumes you have Docker Desktop installed, Python installed, and a GitHub account ready to go. You will also need a free CoinCap API account (you will set this up in Step 1).

Install Astro CLI and Initialize Project

Every data pipeline needs an orchestrator. The orchestrator decides when tasks run, in what order, and what happens if something fails.

In this project, you will use Apache Airflow as your orchestrator, managed through the Astro CLI. The Astro CLI gives you a local Airflow environment running inside Docker containers. In this step, you will install the CLI, scaffold a project, and verify Airflow is running.

In this step, get ready to:

• Install the Astro CLI.
• Initialize a new Astro project and verify it starts.
• Configure the project structure for the pipeline.

Install the Astro CLI

The Astro CLI is the tool that creates and runs Airflow projects locally. Since you already have Docker Desktop installed, you can skip the bundled Podman dependency that the installer includes by default.

• Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
• Install the Astro CLI by running the command for your operating system:

macOS

• Install the Astro CLI using Homebrew with the Docker-friendly tap by running these commands:

brew tap astronomer/tap
brew install astronomer/tap/astro --without-podman

Why --without-podman?

Starting with version 1.32.0, the Astro CLI installs Podman by default as its container engine. Since you already have Docker Desktop, passing --without-podman tells the installer to skip Podman and use your existing Docker setup.

Windows

• Install the Astro CLI using winget by running this command:

winget install -e --id Astronomer.Astro --skip-dependencies

Why --skip-dependencies?

By default, winget installs Podman alongside the Astro CLI as a container engine. Since you already have Docker Desktop, passing --skip-dependencies tells the installer to skip Podman and use your existing Docker setup.

• Verify the installation by running:

astro version

✔️ I see a version number

The Astro CLI is installed and ready to use. Move on to the next substep.

ⓧ Command not found

The CLI was not added to your system PATH. Try the following:

• Close your terminal completely and reopen it, then run astro version again.
• On macOS, if Homebrew is not configured, follow the instructions printed after brew install to add it to your PATH.
• On Windows, restart PowerShell as Administrator and try the winget install command again.

If you are still stuck, help me troubleshoot the Astro CLI installation.

Initialize the Astro project

The Astro CLI scaffolds a complete Airflow project structure in an empty directory. This includes folders for DAGs, plugins, and configuration files.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir crypto-pipeline
cd crypto-pipeline

• Initialize the Astro project by running:

astro dev init

What does this command do?

This scaffolds a full Airflow project inside crypto-pipeline/. It creates:

• dags/ for your pipeline DAG files.
• include/ for extra scripts and data files.
• plugins/ for custom Airflow plugins.
• Dockerfile specifying the Astro Runtime image.
• .env for local environment variables.
• requirements.txt for Python package dependencies.

• Confirm the project files exist by running:

ls

You should see Dockerfile, dags, include, plugins, requirements.txt, .env, and other files in the output.

• Now verify that the project starts correctly by running:

astro dev start

What happens when you run this?

The Astro CLI builds a Docker image from your project and spins up four containers: a Postgres metadata database, the Airflow Webserver, the Scheduler, and the Triggerer. This may take a few minutes on the first run as it downloads the Astro Runtime image.

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

You should see the Airflow UI with an example DAG listed.

Don't see the Airflow UI?

• Make sure Docker Desktop is running. The Astro CLI requires Docker to spin up its containers.
• Wait a minute and refresh the page. The webserver can take time to become healthy on the first start.
• If you see a port conflict error, another service may be using port 8080. Stop that service and try again.

Still stuck? Help me troubleshoot astro dev start.

• Now that you have confirmed Airflow works, stop the environment so you can configure the project before starting it again in the next step:

astro dev stop

Configure the project structure

The default scaffold is almost ready. You need to add a subdirectory for your Python scripts, update the Dockerfile to use the latest runtime, and add your CoinCap API key to the environment file.

• Create the scripts directory inside include/ by running:

mkdir -p include/scripts
touch include/scripts/.gitkeep

Why a .gitkeep file?

Git does not track empty directories. Adding a .gitkeep file ensures the include/scripts/ folder is committed to your repository even before you add real scripts to it.

• Open the Dockerfile in a text editor. Replace its entire contents with:

FROM quay.io/astronomer/astro-runtime:latest

What does this line do?

This tells Docker to build your Airflow image from the latest Astro Runtime. Astro Runtime is a production-ready distribution of Apache Airflow maintained by Astronomer. Using :latest keeps you on the newest stable release.

• Open the .env file in your text editor. Replace its entire contents with:

COINCAP_API_KEY=your_api_key_here

• Replace your_api_key_here with the actual API key you got from your CoinCap dashboard at pro.coincap.io/dashboard.
• Save both files.

Keep your API key safe

The .env file is automatically excluded from Docker image builds. Never commit it to a public repository. Later, you will add it to .gitignore as an extra safeguard.

• Confirm your project structure looks correct by running:

ls include/scripts

You should see .gitkeep in the output.

Missing .gitkeep or scripts folder?

• Make sure you ran the mkdir and touch commands from inside the crypto-pipeline/ directory, not from your Desktop.
• Run pwd to confirm your current location ends with crypto-pipeline.

Still having trouble? Help me fix my project structure.

✔️ Awesome, I've got everything!

Great. Double check that you have saved both Dockerfile and .env.

ⓧ I'd like to double check the full code

Here is what each file should contain at this point:

Dockerfile:

FROM quay.io/astronomer/astro-runtime:latest

.env:

COINCAP_API_KEY=your_api_key_here

include/scripts/.gitkeep:

This file is empty. It exists only to keep the directory tracked in Git.

requirements.txt:

This file is empty for now. You will add Python dependencies in later steps.

Your Astro project is scaffolded and verified. Next up, you will add the Docker infrastructure for MinIO, Spark, and Metabase to power the full data pipeline.

Set Up Docker Infrastructure

Your Astro project is running, but right now it only has Airflow. A real data pipeline needs storage for raw and processed data, a processing engine for transformations, and a visualization layer for insights.

In this step, you will add three critical infrastructure services to your project using Docker. MinIO provides S3-compatible object storage. Apache Spark handles data transformations at scale. Metabase visualizes your warehouse data in dashboards.

In this step, get ready to:

• Create a Docker Compose override file defining MinIO, Spark, and Metabase services.
• Start the full infrastructure stack and verify all services are accessible.
• Create MinIO buckets for the bronze and silver data layers.

Create the Docker Compose override file

The Astro CLI supports a docker-compose.override.yml file that lets you add custom services alongside Airflow. This is how you extend the default stack without modifying Astro's internal configuration.

Before creating the override file, you need a custom Metabase image. The default Metabase Docker image does not include the DuckDB driver, so you will build a custom image that bundles it in.

• Create the include/metabase/ directory inside your crypto-pipeline folder by running this command:

mkdir include/metabase

• Confirm the directory exists by running:

ls include/

• You should see metabase and scripts in the output.

• Create a file called Dockerfile inside include/metabase/ and paste the following content:

FROM metabase/metabase:latest
ADD https://github.com/MotherDuck-Open-Source/metabase_duckdb_driver/releases/download/0.2.3/duckdb.metabase-driver.jar /plugins/
RUN chmod 744 /plugins/duckdb.metabase-driver.jar

What does this Dockerfile do?

This Dockerfile starts from the official Metabase image and adds the DuckDB community driver JAR file into the /plugins/ directory. Metabase automatically loads any driver JARs it finds in that folder on startup.

The chmod 744 ensures the JAR file has the correct read permissions inside the container.

• Save include/metabase/Dockerfile.

Don't see the Dockerfile?

Make sure you created the include/metabase/ directory first with mkdir. If the directory does not exist, your file creation will fail.

Double-check that your terminal is inside the crypto-pipeline folder. Run pwd to confirm.

Help me troubleshoot my Metabase Dockerfile setup.

Now create the Docker Compose override file that defines all three infrastructure services.

• Create a file called docker-compose.override.yml in the root of your crypto-pipeline folder and paste the following content:

version: '3'
services:
  minio:
    image: minio/minio
    ports:
      - '9000:9000'
      - '9001:9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ':9001'
    volumes:
      - minio-data:/data
  spark-master:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=master
    ports:
      - '7077:7077'
      - '8081:8080'
  spark-worker:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
    depends_on:
      - spark-master
  metabase:
    image: metabase-duckdb:latest
    build:
      context: ./include/metabase
    ports:
      - '3000:3000'
    volumes:
      - metabase-data:/metabase-data
volumes:
  minio-data:
  metabase-data:

What does this configuration define?

• minio: Runs MinIO on port 9000 (API) and 9001 (web console). Uses minioadmin as both the username and password for local development.
• spark-master and spark-worker: A two-node Spark cluster using the bitnami/spark image. The master listens on port 7077 and exposes its UI on port 8081.
• metabase: Builds from your custom Dockerfile and exposes the Metabase dashboard on port 3000.
• volumes: Named Docker volumes persist MinIO data and Metabase configuration across restarts.

• Save docker-compose.override.yml.

Finally, update your .env file so Airflow tasks can connect to MinIO later.

• Open the .env file in your crypto-pipeline folder.
• Find this line:

COINCAP_API_KEY=your_api_key_here

• Add the following three lines below it:

MINIO_ENDPOINT=http://minio:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin

Why use http://minio:9000?

Inside Docker, containers communicate using service names as hostnames. The Airflow container reaches MinIO at minio:9000 because Docker networking resolves the service name minio to the correct container IP.

From your browser, you use localhost:9001 instead. These are two different network paths to the same service.

• Save the .env file.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files before continuing.

␧ I'd like to double check the full code

Here is the full docker-compose.override.yml:

version: '3'
services:
  minio:
    image: minio/minio
    ports:
      - '9000:9000'
      - '9001:9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ':9001'
    volumes:
      - minio-data:/data
  spark-master:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=master
    ports:
      - '7077:7077'
      - '8081:8080'
  spark-worker:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
    depends_on:
      - spark-master
  metabase:
    image: metabase-duckdb:latest
    build:
      context: ./include/metabase
    ports:
      - '3000:3000'
    volumes:
      - metabase-data:/metabase-data
volumes:
  minio-data:
  metabase-data:

Here is the full include/metabase/Dockerfile:

FROM metabase/metabase:latest
ADD https://github.com/MotherDuck-Open-Source/metabase_duckdb_driver/releases/download/0.2.3/duckdb.metabase-driver.jar /plugins/
RUN chmod 744 /plugins/duckdb.metabase-driver.jar

Here is the full .env:

COINCAP_API_KEY=your_api_key_here
MINIO_ENDPOINT=http://minio:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin

Start the full stack and verify services

Your Airflow environment is currently running without the new services. You need to stop it and restart so the Astro CLI picks up the new docker-compose.override.yml file.

• Stop the running Airflow environment by running:

astro dev stop

• Start the full stack (Airflow plus your new services) by running:

astro dev start

What happens during startup?

The Astro CLI detects your docker-compose.override.yml and merges it with its internal Docker Compose configuration. It builds the custom Metabase image, pulls the MinIO and Spark images, and starts everything together.

This may take a few minutes on the first run while Docker downloads the images.

• Once the command completes, verify all services are running by opening these URLs in your browser:
• Airflow UI: http://localhost:8080 (login: admin / admin).
• MinIO Console: http://localhost:9001 (login: minioadmin / minioadmin).
• Spark Master UI: http://localhost:8081.
• Metabase: http://localhost:3000.

Services not starting?

If you see port conflict errors, make sure no other applications are using ports 9000, 9001, 8081, or 3000. Stop any conflicting services before retrying.

If the Metabase image fails to build, check that your include/metabase/Dockerfile is saved correctly. A typo in the URL will cause the build to fail.

If Docker runs out of memory, increase the memory allocation in Docker Desktop settings to at least 6 GB. The Spark containers require significant memory.

Help me troubleshoot why my stack is not starting.

Create MinIO buckets for bronze and silver layers

Your pipeline will store data in two buckets following the Medallion Architecture. The bronze bucket holds raw API responses. The silver bucket holds cleaned and transformed data.

• Open the MinIO Console at http://localhost:9001 in your browser.
• Log in with username minioadmin and password minioadmin.
• Click Create Bucket.
• Enter bronze as the bucket name and click Create Bucket to confirm.
• Click Create Bucket again to start a second bucket.
• Enter silver as the bucket name and click Create Bucket to confirm.

You should now see both bronze and silver listed in the MinIO Console's bucket view.

What is the Medallion Architecture?

The Medallion Architecture organizes data in layers of increasing quality. Bronze holds raw, unmodified data straight from the source. Silver holds cleaned and validated data. Gold holds business-ready, aggregated data.

This pattern makes debugging easier because you can always trace back to the original raw data in bronze.

Can't access the MinIO Console?

Make sure you are navigating to http://localhost:9001 (the console port), not http://localhost:9000 (the API port). The API port returns XML responses, not a web interface.

If the page does not load at all, run docker ps in your terminal to verify the MinIO container is running.

Help me access the MinIO Console.

Your local data platform is up and running. Next up, you will write the extraction script that pulls live cryptocurrency data from the CoinCap API into your bronze bucket.

Extract Crypto Data to Bronze Layer

Your infrastructure is running and your MinIO buckets are ready. Now it is time to build the first real stage of your pipeline: extracting live cryptocurrency data and landing it in the bronze layer.

The bronze layer stores raw, untransformed data exactly as it arrives from the source. You will write a Python script that calls the CoinCap API v3, then wire it into an Airflow DAG so the extraction runs on a schedule.

In this step, get ready to:

• Write a Python extraction script that pulls crypto asset data from CoinCap and uploads raw JSON to MinIO.
• Create an Airflow DAG that orchestrates the extraction on a daily schedule.
• Trigger the DAG manually and verify raw data lands in the bronze bucket.

Write the extraction script

The extraction script is the entry point for data flowing into your pipeline. It calls the CoinCap API, receives a JSON payload of cryptocurrency asset data, and pushes that raw response into MinIO using the S3-compatible boto3 client.

• Add the Python packages the extraction script needs by opening requirements.txt in your crypto-pipeline project and pasting the following:

boto3
requests

What do these packages do?

The boto3 library is the AWS SDK for Python. Because MinIO implements the S3 API, boto3 can talk to MinIO without any code changes.

The requests library handles HTTP calls to the CoinCap API.

• Create a new file at include/scripts/extract.py inside your crypto-pipeline project. This is where your extraction logic lives.
• Add the imports and configuration block by pasting the following into extract.py:

import os
import json
import boto3
import requests
from datetime import date

# Configuration pulled from environment variables set in .env
MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
COINCAP_API_KEY = os.getenv('COINCAP_API_KEY')
BRONZE_BUCKET = 'bronze'

What does this code do?

Each os.getenv() call reads a value from the environment variables defined in your .env file. The fallback values (like 'minioadmin') ensure the script works even if a variable is missing.

The BRONZE_BUCKET constant points to the MinIO bucket you created in the previous step.

• Add the extraction function below the configuration block in extract.py:

def extract_assets():
    # Call CoinCap API v3 with bearer token authentication
    url = 'https://rest.coincap.io/v3/assets'
    headers = {'Authorization': f'Bearer {COINCAP_API_KEY}'}
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    data = response.json()
    # Upload raw JSON to MinIO bronze bucket
    s3 = boto3.client(
        's3',
        endpoint_url=MINIO_ENDPOINT,
        aws_access_key_id=MINIO_ACCESS_KEY,
        aws_secret_access_key=MINIO_SECRET_KEY
    )
    today = date.today().isoformat()
    key = f'assets/{today}/assets.json'
    s3.put_object(Bucket=BRONZE_BUCKET, Key=key, Body=json.dumps(data))
    print(f'Uploaded raw data to bronze/{key}')
    return key

if __name__ == '__main__':
    extract_assets()

What does this function do?

The extract_assets() function sends a GET request to the CoinCap API v3 assets endpoint. It uses bearer token authentication via the Authorization header.

After receiving the response, it creates a boto3 S3 client pointed at MinIO and uploads the raw JSON to a date-partitioned path like assets/2026-05-21/assets.json. Partitioning by date makes it easy to track when each extraction ran.

The if __name__ == '__main__' block lets you test the script standalone outside Airflow if needed.

• Save extract.py.

Not sure where to create the file?

The file path is include/scripts/extract.py relative to your crypto-pipeline project root. If the scripts folder does not exist yet inside include/, create it first.

Help me set up the extraction script file structure.

✔️ Awesome, I've got everything!

Great. Make sure you have saved both requirements.txt and extract.py.

ⓧ I'd like to double check the full code

import os
import json
import boto3
import requests
from datetime import date

MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
COINCAP_API_KEY = os.getenv('COINCAP_API_KEY')
BRONZE_BUCKET = 'bronze'

def extract_assets():
    url = 'https://rest.coincap.io/v3/assets'
    headers = {'Authorization': f'Bearer {COINCAP_API_KEY}'}
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    data = response.json()
    s3 = boto3.client(
        's3',
        endpoint_url=MINIO_ENDPOINT,
        aws_access_key_id=MINIO_ACCESS_KEY,
        aws_secret_access_key=MINIO_SECRET_KEY
    )
    today = date.today().isoformat()
    key = f'assets/{today}/assets.json'
    s3.put_object(Bucket=BRONZE_BUCKET, Key=key, Body=json.dumps(data))
    print(f'Uploaded raw data to bronze/{key}')
    return key

if __name__ == '__main__':
    extract_assets()

Create the Airflow DAG

An Airflow DAG (Directed Acyclic Graph) defines the order and schedule of tasks in your pipeline. You will create a DAG that calls your extraction function on a daily schedule.

• Create a new file at dags/crypto_pipeline.py inside your crypto-pipeline project and paste the following:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys

# Add include/scripts to the Python path so Airflow can find extract.py
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(
        task_id='extract_assets',
        python_callable=extract_assets
    )

What does this DAG do?

The sys.path.insert line adds the include/scripts directory to the Python path inside the Airflow container. This lets you import extract_assets from your script.

The PythonOperator wraps your function as an Airflow task. When this task runs, it calls extract_assets() and the data flows from CoinCap into your bronze bucket.

Setting catchup=False prevents Airflow from backfilling every day since the start_date. The schedule='@daily' means the DAG will trigger once per day going forward.

• Save crypto_pipeline.py.

Why PythonOperator instead of the @task decorator?

The PythonOperator pattern keeps the extraction logic in a separate file (extract.py) rather than inline in the DAG. This makes it easier to test independently, reuse across multiple DAGs, and keep the DAG file short and readable.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the DAG file.

ⓧ I'd like to double check the full code

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(
        task_id='extract_assets',
        python_callable=extract_assets
    )

Test the extraction by triggering the DAG

You have written the code, but the Airflow container does not yet have boto3 and requests installed. Changes to requirements.txt require a full environment restart to take effect.

• Rebuild and restart your Astro environment by running this command in your terminal:

astro dev restart

Why do I need to restart?

The Astro CLI rebuilds the Docker image when you run astro dev restart. DAG code changes are picked up automatically, but changes to requirements.txt, packages.txt, or the Dockerfile require a restart so the new packages get installed into the container image.

Restart taking a long time?

The first restart after adding packages may take a minute or two while Docker downloads and installs boto3 and requests. Subsequent restarts will be faster thanks to Docker layer caching.

Help me troubleshoot my Astro CLI restart.

• Once the restart completes, open the Airflow UI at http://localhost:8080 in your browser.
• Find crypto_pipeline in the DAG list.
• Unpause the DAG by clicking the toggle switch next to its name.
• Click the play button (trigger icon) to manually trigger a DAG run.

What should I see?

After triggering, you will see a DAG run appear with a running state (light green circle). Within a few seconds it should complete and show a dark green circle indicating success.

If you click the DAG run, then click the extract_assets task and open its logs, you should see the message Uploaded raw data to bronze/assets/<today's date>/assets.json.

• Verify the data landed in MinIO by opening http://localhost:9001 in your browser.
• Click the bronze bucket.
• Navigate into the assets/ folder, then today's date folder.
• You should see assets.json listed. Click it to preview the raw JSON from CoinCap.

DAG not showing up in the Airflow UI?

Make sure you saved crypto_pipeline.py in the dags/ folder (not include/). Airflow only scans the dags/ directory for DAG files.

If the DAG appears but has an import error icon, check the Airflow logs. A common cause is a typo in the import path or missing packages (which means the restart did not pick up your requirements.txt changes).

Help me figure out why my DAG is not appearing in Airflow.

DAG run failed with an authentication error?

If you see a 401 or 403 error in the task logs, double-check that your COINCAP_API_KEY in the .env file is a valid API key from your CoinCap dashboard at pro.coincap.io/dashboard. The free DEMO tier provides 4,000 credits per month which is more than enough for this project.

Help me debug my CoinCap API authentication error.

Your extraction pipeline is live. Raw cryptocurrency data from CoinCap is now landing in your bronze bucket, date-partitioned and ready for transformation. Next up, you will write a PySpark job to clean this raw JSON and produce structured Parquet files in the silver layer.

Transform Data with PySpark to Silver Layer

Your bronze layer now holds raw JSON straight from the CoinCap API. Raw data is useful for auditing, but it is messy for analytics.

In this step, you will write a PySpark job that reads the raw JSON from MinIO, cleans and flattens it, casts columns to proper types, and writes the output as Parquet files into your silver bucket. This is the "T" in ETL.

In this step, get ready to:

• Write a PySpark transformation script that reads bronze JSON and writes silver Parquet.
• Add the transformation as a task in your Airflow DAG with proper dependencies.
• Verify Parquet files appear in the MinIO silver bucket.

Write the PySpark transformation script

PySpark uses the S3A filesystem connector to read and write directly to MinIO. Your script will create a SparkSession configured with MinIO credentials, read the raw JSON, cast numeric fields, drop rows missing critical data, and write clean Parquet.

• Create a new file called transform.py inside include/scripts/ by pasting the following imports and configuration:

import os
from datetime import date
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType, LongType

# MinIO connection settings (same credentials as extract.py)
MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')

What does this code do?

The imports pull in PySpark's SparkSession (the entry point for Spark operations), col for column references, and DoubleType / LongType for casting string values into real numeric types.

The three environment variables reuse the same MinIO credentials your extract script already uses. Keeping them consistent means both scripts talk to the same storage backend.

• Add the transform_assets() function below the configuration constants:

def transform_assets():
    # Use today's date to locate the correct bronze partition
    today = date.today().isoformat()

    # Create a Spark session configured to read/write MinIO via S3A
    spark = SparkSession.builder \
        .appName('CryptoTransform') \
        .config('spark.hadoop.fs.s3a.endpoint', MINIO_ENDPOINT) \
        .config('spark.hadoop.fs.s3a.access.key', MINIO_ACCESS_KEY) \
        .config('spark.hadoop.fs.s3a.secret.key', MINIO_SECRET_KEY) \
        .config('spark.hadoop.fs.s3a.path.style.access', 'true') \
        .getOrCreate()

    # Read raw JSON from the bronze layer
    df = spark.read.json(f's3a://bronze/assets/{today}/assets.json')

    # Flatten nested 'data' field if present in the API response
    assets_df = df.select('data.*') if 'data' in df.columns else df

    # Select and cast columns to proper numeric types
    clean_df = assets_df.select(
        col('id'),
        col('rank').cast(LongType()),
        col('symbol'),
        col('name'),
        col('priceUsd').cast(DoubleType()),
        col('marketCapUsd').cast(DoubleType()),
        col('volumeUsd24Hr').cast(DoubleType()),
        col('changePercent24Hr').cast(DoubleType())
    ).dropna(subset=['id', 'priceUsd'])

    # Write cleaned data as Parquet to the silver layer
    clean_df.write.mode('overwrite').parquet(f's3a://silver/assets/{today}/')
    print(f'Wrote Parquet to silver/assets/{today}/')
    spark.stop()

if __name__ == '__main__':
    transform_assets()

What does this function do?

• SparkSession.builder creates a local Spark session pointed at MinIO. The fs.s3a.path.style.access config tells Spark to use path-style URLs instead of virtual-hosted, which MinIO requires.
• df.select('data.*') flattens the nested data array from the CoinCap JSON response into individual rows.
• The .cast() calls convert price and volume strings into proper numeric types so downstream queries can aggregate them.
• .dropna(subset=['id', 'priceUsd']) removes any rows missing the two most critical fields.
• The final .write.mode('overwrite').parquet() writes the cleaned DataFrame as columnar Parquet files, which are much faster to query than raw JSON.

• Save transform.py.

You won't see a change yet. The next substep wires this into the Airflow DAG so it actually runs.

Seeing a syntax error when you save?

• Make sure the backslashes at the end of each .config() line have no trailing spaces. Python line continuations break with trailing whitespace.
• Check that the indentation inside the function uses consistent spaces (4 spaces per level).

Help me fix a syntax error in my PySpark transform script.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

import os
from datetime import date
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType, LongType

MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')

def transform_assets():
    today = date.today().isoformat()
    spark = SparkSession.builder \
        .appName('CryptoTransform') \
        .config('spark.hadoop.fs.s3a.endpoint', MINIO_ENDPOINT) \
        .config('spark.hadoop.fs.s3a.access.key', MINIO_ACCESS_KEY) \
        .config('spark.hadoop.fs.s3a.secret.key', MINIO_SECRET_KEY) \
        .config('spark.hadoop.fs.s3a.path.style.access', 'true') \
        .getOrCreate()
    df = spark.read.json(f's3a://bronze/assets/{today}/assets.json')
    assets_df = df.select('data.*') if 'data' in df.columns else df
    clean_df = assets_df.select(
        col('id'),
        col('rank').cast(LongType()),
        col('symbol'),
        col('name'),
        col('priceUsd').cast(DoubleType()),
        col('marketCapUsd').cast(DoubleType()),
        col('volumeUsd24Hr').cast(DoubleType()),
        col('changePercent24Hr').cast(DoubleType())
    ).dropna(subset=['id', 'priceUsd'])
    clean_df.write.mode('overwrite').parquet(f's3a://silver/assets/{today}/')
    print(f'Wrote Parquet to silver/assets/{today}/')
    spark.stop()

if __name__ == '__main__':
    transform_assets()

Add the PySpark transformation task to the DAG

The transform script is ready, but Airflow does not know about it yet. You need to register it as a task in the DAG and declare that it depends on the extraction finishing first.

• Open requirements.txt in your crypto-pipeline project root.
• Find this line:

boto3
requests

• Add pyspark on a new line below requests, so the file now looks like this:

boto3
requests
pyspark

• Save requirements.txt.

Why add pyspark to requirements.txt?

The Astro CLI builds the Airflow container image from this file. Adding pyspark here ensures the PySpark library is installed inside the Airflow container where your DAG tasks run.

• Switch to dags/crypto_pipeline.py.
• Find this import line near the top of the file:

from extract import extract_assets

• Add this import directly below it:

from transform import transform_assets

• Scroll down to the end of the DAG block, below the existing extract_task definition. Add the transform task and the dependency chain:

    # Transform raw JSON into clean Parquet in the silver layer
    transform_task = PythonOperator(
        task_id='transform_assets',
        python_callable=transform_assets
    )

    # Run extract first, then transform
    extract_task >> transform_task

What does the >> operator do?

In Airflow, the >> operator sets a dependency between tasks. extract_task >> transform_task means "run extract_task first, and only start transform_task after it succeeds."

• Save dags/crypto_pipeline.py.

Not sure where to paste the new code?

The transform_task definition goes inside the with DAG(...) as dag: block, after the extract_task definition. Both tasks must be indented at the same level (one level inside the with block).

Help me figure out where to place transform_task in my DAG file.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before continuing.

ⓧ I'd like to double check the full code

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets
from transform import transform_assets

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(
        task_id='extract_assets',
        python_callable=extract_assets
    )
    transform_task = PythonOperator(
        task_id='transform_assets',
        python_callable=transform_assets
    )
    extract_task >> transform_task

• Rebuild your Astro environment to install the new pyspark dependency by running:

astro dev restart

Why restart instead of just saving?

Changes to requirements.txt require rebuilding the Airflow container image. The astro dev restart command stops the environment, rebuilds the image with your new dependency, and starts everything again.

Wait for all containers to start. You should see output confirming the webserver, scheduler, and other services are running.

Restart failing or taking too long?

• Make sure Docker Desktop is still running. The restart needs Docker to rebuild the image.
• If you see a dependency resolution error, check that requirements.txt contains exactly three lines: boto3, requests, pyspark. No version pins or typos.

Help me troubleshoot my astro dev restart failure.

Verify Parquet files in the MinIO silver bucket

Time to run the full pipeline and confirm that clean Parquet files land in your silver bucket.

• Return to the Airflow UI at localhost:8080 in your browser.
• Find the crypto_pipeline DAG in the list.
• Click the play button (trigger icon) on the right side to trigger a manual run.
• Click on the DAG name to open its detail view.
• Watch the Grid view. You should see both extract_assets and transform_assets tasks turn green (success).

What should I see?

Both tasks should complete successfully within a minute or two. The extract task runs first, then the transform task fires automatically because of the >> dependency you set.

• Open the MinIO Console at localhost:9001 in your browser.
• Click on the silver bucket in the left sidebar.
• Navigate into assets/ and then today's date folder.
• You should see one or more .parquet files listed. This confirms PySpark successfully transformed your raw JSON into columnar Parquet.

Transform task failed (red in Airflow)?

• Click on the failed transform_assets task, then click the Log tab. The error message tells you exactly what went wrong.
• If the log says the bronze file was not found, make sure the extract task ran successfully first. The transform reads from today's date partition, so both tasks must run on the same day.
• If you see a connection refused error to MinIO, check that the minio container is running by looking at Docker Desktop.

Help me debug my transform_assets task failure in Airflow.

Your pipeline now has two working stages. Raw JSON flows into bronze, and PySpark transforms it into clean, typed Parquet in silver. Next up, you will load this silver data into DuckDB to create analytical gold tables ready for dashboards.

Load Data into DuckDB Gold Layer

Your silver layer now contains clean, structured Parquet files ready for analytical queries. The final piece of the Medallion Architecture is the gold layer, where data is modeled into purpose-built tables optimized for business questions.

In this step, you will use DuckDB as your analytical warehouse. DuckDB reads Parquet files directly from MinIO via its httpfs extension, so there is no separate data-loading step needed.

In this step, get ready to:

• Write a Python script that connects DuckDB to MinIO and creates gold-layer analytical tables.
• Add the DuckDB loading task to the Airflow DAG as the final step.
• Verify the gold tables are populated by querying DuckDB directly.

Write the DuckDB loading script

The loading script will connect to a DuckDB database file stored locally, configure it to read from MinIO, and create three gold tables. Each table serves a different analytical purpose.

• Create the warehouse directory inside include/ by running this command from your crypto-pipeline folder:

mkdir include/warehouse

• Confirm the directory exists by running:

ls include/

You should see warehouse alongside scripts and metabase in the output.

• Create a new file called load.py inside include/scripts/ and paste the following code:

import os
import duckdb
from datetime import date

# MinIO connection settings (same credentials as other scripts)
MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')

# Path where DuckDB stores its database file
DUCKDB_PATH = '/usr/local/airflow/include/warehouse/crypto.duckdb'

def load_gold_tables():
    today = date.today().isoformat()

    # Connect to the DuckDB database file
    con = duckdb.connect(DUCKDB_PATH)

    # Install and load httpfs so DuckDB can read from MinIO (S3-compatible)
    con.execute('INSTALL httpfs; LOAD httpfs;')

    # Configure S3 settings to point at local MinIO
    con.execute(f"SET s3_endpoint='{MINIO_ENDPOINT.replace('http://', '')}'; SET s3_access_key_id='{MINIO_ACCESS_KEY}'; SET s3_secret_access_key='{MINIO_SECRET_KEY}'; SET s3_use_ssl=false; SET s3_url_style='path';")

    # Gold table 1: Asset rankings ordered by market cap rank
    con.execute(f"CREATE OR REPLACE TABLE gold_asset_rankings AS SELECT id, rank, symbol, name, priceUsd, marketCapUsd FROM read_parquet('s3://silver/assets/{today}/*.parquet') ORDER BY rank ASC;")

    # Gold table 2: Aggregated market summary stats
    con.execute(f"CREATE OR REPLACE TABLE gold_market_summary AS SELECT COUNT(*) as total_assets, SUM(marketCapUsd) as total_market_cap, SUM(volumeUsd24Hr) as total_volume_24h, AVG(changePercent24Hr) as avg_change_24h FROM read_parquet('s3://silver/assets/{today}/*.parquet');")

    # Gold table 3: Price aggregations ordered by 24h volume
    con.execute(f"CREATE OR REPLACE TABLE gold_price_aggregations AS SELECT id, symbol, name, priceUsd, marketCapUsd, volumeUsd24Hr, changePercent24Hr FROM read_parquet('s3://silver/assets/{today}/*.parquet') ORDER BY volumeUsd24Hr DESC;")

    print('Gold tables created: gold_asset_rankings, gold_market_summary, gold_price_aggregations')
    con.close()

if __name__ == '__main__':
    load_gold_tables()

What does this code do?

This script connects DuckDB to MinIO using the httpfs extension, which lets DuckDB read files from S3-compatible storage as if they were local.

• gold_asset_rankings lists every crypto asset ordered by rank, useful for a top-10 leaderboard.
• gold_market_summary aggregates the entire market into a single row of totals and averages, useful for headline stats.
• gold_price_aggregations lists every asset ordered by 24-hour trading volume, useful for spotting the most actively traded coins.

CREATE OR REPLACE TABLE ensures the script is idempotent. Running it multiple times on the same day overwrites old data without errors.

• Save include/scripts/load.py.

File not saving in the right place?

Make sure you created load.py inside include/scripts/ (the same folder as extract.py and transform.py). The DAG imports from this exact directory.

help me debug my project file structure for the DuckDB loading script

Add the load task to the Airflow DAG

With the loading script written, you need to register it as a task in the DAG and add duckdb to the project dependencies so the Airflow container can import it.

• Open requirements.txt in your crypto-pipeline folder and find this line:

pyspark

• Add duckdb on a new line below pyspark, so the file now looks like this:

boto3
requests
pyspark
duckdb

• Save requirements.txt.
• Now open dags/crypto_pipeline.py and replace the entire contents with the following updated DAG:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets
from transform import transform_assets
from load import load_gold_tables

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(task_id='extract_assets', python_callable=extract_assets)
    transform_task = PythonOperator(task_id='transform_assets', python_callable=transform_assets)
    load_task = PythonOperator(task_id='load_gold_tables', python_callable=load_gold_tables)
    extract_task >> transform_task >> load_task

What changed in the DAG?

• A new import from load import load_gold_tables brings in the function you just wrote.
• A new load_task PythonOperator calls load_gold_tables to create the gold tables.
• The dependency chain is now extract_task >> transform_task >> load_task, meaning DuckDB loading only runs after PySpark finishes transforming.

• Save dags/crypto_pipeline.py.

DAG not appearing in Airflow UI?

If the DAG disappears or shows an import error in the Airflow UI, check that load.py is in include/scripts/ (not dags/). Also make sure the function name load_gold_tables matches exactly between the import and the script.

help me fix my DAG import error for the DuckDB load task

Verify the gold tables are populated

Since you updated requirements.txt, you need to restart the Astro environment so it installs duckdb inside the Airflow container.

• Restart your Astro environment by running:

astro dev restart

Why restart instead of just restarting the scheduler?

Changes to requirements.txt require a full image rebuild. The astro dev restart command rebuilds the Docker image with duckdb installed, then restarts all containers.

• Wait for the restart to finish. You will see a message confirming all containers are running.
• Open the Airflow UI at localhost:8080 in your browser.
• Find the crypto_pipeline DAG and click the play button to trigger a manual run.
• Wait for all three tasks (extract_assets, transform_assets, load_gold_tables) to show green (success).

Now verify the gold tables by querying DuckDB directly inside the Airflow container.

• Open a bash shell inside the scheduler container by running:

astro dev bash

• Once inside the container, query the gold tables by running:

python -c "import duckdb; con = duckdb.connect('/usr/local/airflow/include/warehouse/crypto.duckdb', read_only=True); print(con.execute('SELECT symbol, name, rank FROM gold_asset_rankings LIMIT 5').fetchall()); con.close()"

You should see a list of the top 5 cryptocurrencies by rank, such as BTC (Bitcoin) at rank 1 and ETH (Ethereum) at rank 2.

• Exit the container shell by typing:

exit

Seeing "Table does not exist" or empty results?

If DuckDB reports that gold_asset_rankings does not exist, the load_gold_tables task may not have run successfully. Check its logs in the Airflow UI by clicking on the task and selecting **Log**.

If you see an error about httpfs, confirm the MinIO service is running. The MINIO_ENDPOINT inside the container must resolve to http://minio:9000 since all services share the same Docker network.

help me debug why my DuckDB gold tables are empty or missing

✔️ Awesome, I've got everything!

Great. Double check you have saved both include/scripts/load.py and dags/crypto_pipeline.py.

ⓧ I'd like to double check the full code

Here is the complete code for each file at this point:

import os
import duckdb
from datetime import date

MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
DUCKDB_PATH = '/usr/local/airflow/include/warehouse/crypto.duckdb'

def load_gold_tables():
    today = date.today().isoformat()
    con = duckdb.connect(DUCKDB_PATH)
    con.execute('INSTALL httpfs; LOAD httpfs;')
    con.execute(f"SET s3_endpoint='{MINIO_ENDPOINT.replace('http://', '')}'; SET s3_access_key_id='{MINIO_ACCESS_KEY}'; SET s3_secret_access_key='{MINIO_SECRET_KEY}'; SET s3_use_ssl=false; SET s3_url_style='path';")
    con.execute(f"CREATE OR REPLACE TABLE gold_asset_rankings AS SELECT id, rank, symbol, name, priceUsd, marketCapUsd FROM read_parquet('s3://silver/assets/{today}/*.parquet') ORDER BY rank ASC;")
    con.execute(f"CREATE OR REPLACE TABLE gold_market_summary AS SELECT COUNT(*) as total_assets, SUM(marketCapUsd) as total_market_cap, SUM(volumeUsd24Hr) as total_volume_24h, AVG(changePercent24Hr) as avg_change_24h FROM read_parquet('s3://silver/assets/{today}/*.parquet');")
    con.execute(f"CREATE OR REPLACE TABLE gold_price_aggregations AS SELECT id, symbol, name, priceUsd, marketCapUsd, volumeUsd24Hr, changePercent24Hr FROM read_parquet('s3://silver/assets/{today}/*.parquet') ORDER BY volumeUsd24Hr DESC;")
    print('Gold tables created: gold_asset_rankings, gold_market_summary, gold_price_aggregations')
    con.close()

if __name__ == '__main__':
    load_gold_tables()

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets
from transform import transform_assets
from load import load_gold_tables

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(task_id='extract_assets', python_callable=extract_assets)
    transform_task = PythonOperator(task_id='transform_assets', python_callable=transform_assets)
    load_task = PythonOperator(task_id='load_gold_tables', python_callable=load_gold_tables)
    extract_task >> transform_task >> load_task

boto3
requests
pyspark
duckdb

Your pipeline now covers the full Medallion Architecture. Raw data lands in bronze, gets cleaned into silver Parquet, and is modeled into gold tables inside DuckDB. Next up, you will add data quality checks between each stage to catch problems before they propagate downstream.

Add Data Quality Checks

Your pipeline extracts, transforms, and loads data end-to-end. But what happens if the API returns garbage data, the transformation produces empty files, or the gold tables silently lose rows?

In production, data teams add quality checks between every stage of a pipeline. These checks catch problems early, before bad data propagates downstream and corrupts dashboards or reports.

In this step, get ready to:

• Write data quality validation functions for each pipeline layer.
• Add quality check tasks between each stage in the DAG.
• Configure the DAG to halt execution if any check fails.

Implement data quality validation functions

You will create a new script that validates data at each layer of the Medallion Architecture. Bronze checks confirm the raw API response is healthy. Silver checks validate the cleaned Parquet data. Gold checks ensure the warehouse tables are populated.

• Create a new file called quality_checks.py inside the include/scripts/ folder (the same folder as your other scripts).
• Add the imports and connection setup by pasting this code:

import os
import json
import boto3
import duckdb
from datetime import date
from airflow.exceptions import AirflowException

# MinIO connection settings (same as other scripts)
MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
DUCKDB_PATH = '/usr/local/airflow/include/warehouse/crypto.duckdb'

def check_bronze_quality():
    today = date.today().isoformat()
    # Connect to MinIO and read today's raw JSON
    s3 = boto3.client('s3', endpoint_url=MINIO_ENDPOINT, aws_access_key_id=MINIO_ACCESS_KEY, aws_secret_access_key=MINIO_SECRET_KEY)
    obj = s3.get_object(Bucket='bronze', Key=f'assets/{today}/assets.json')
    data = json.loads(obj['Body'].read())
    records = data.get('data', data)
    # Check minimum row count threshold
    if len(records) < 10:
        raise AirflowException(f'Bronze quality check failed: only {len(records)} records found')
    # Check for required fields in a sample of records
    for record in records[:5]:
        if not record.get('id') or not record.get('priceUsd'):
            raise AirflowException('Bronze quality check failed: missing required fields')
    print(f'Bronze quality check passed: {len(records)} records')

What does this code do?

The check_bronze_quality function reads today's raw JSON from the bronze bucket and runs two validations.

• Row count threshold: if fewer than 10 records came back, something is wrong with the API response.
• Null checks: it samples the first 5 records and confirms id and priceUsd are present.
• If either check fails, the function raises AirflowException which marks the task as failed in the DAG.

• Now add the silver layer quality check function below the bronze function:

def check_silver_quality():
    today = date.today().isoformat()
    # Connect DuckDB to MinIO via httpfs to read Parquet
    con = duckdb.connect()
    con.execute('INSTALL httpfs; LOAD httpfs;')
    con.execute(f"SET s3_endpoint='{MINIO_ENDPOINT.replace('http://', '')}'; SET s3_access_key_id='{MINIO_ACCESS_KEY}'; SET s3_secret_access_key='{MINIO_SECRET_KEY}'; SET s3_use_ssl=false; SET s3_url_style='path';")
    # Row count check
    count = con.execute(f"SELECT COUNT(*) FROM read_parquet('s3://silver/assets/{today}/*.parquet')").fetchone()[0]
    if count < 10:
        raise AirflowException(f'Silver quality check failed: only {count} rows')
    # Price range validation - no nulls or negative prices allowed
    nulls = con.execute(f"SELECT COUNT(*) FROM read_parquet('s3://silver/assets/{today}/*.parquet') WHERE priceUsd IS NULL OR priceUsd <= 0").fetchone()[0]
    if nulls > 0:
        raise AirflowException(f'Silver quality check failed: {nulls} rows with invalid priceUsd')
    print(f'Silver quality check passed: {count} rows')
    con.close()

What does this code do?

The check_silver_quality function uses DuckDB's httpfs extension to read the Parquet files directly from MinIO without downloading them.

• Schema validation via row count: confirms the transformation produced at least 10 rows.
• Range check on priceUsd: catches any null or non-positive prices that would corrupt downstream analytics.

• Finally, add the gold layer quality check below the silver function:

def check_gold_quality():
    # Connect to the DuckDB warehouse file
    con = duckdb.connect(DUCKDB_PATH)
    # Verify each gold table has data
    for table in ['gold_asset_rankings', 'gold_market_summary', 'gold_price_aggregations']:
        count = con.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0]
        if count == 0:
            raise AirflowException(f'Gold quality check failed: {table} is empty')
    print('Gold quality check passed')
    con.close()

What does this code do?

The check_gold_quality function connects to the crypto.duckdb warehouse and loops over all three gold tables. If any table is empty, it raises an exception to halt the pipeline.

• Save quality_checks.py.
• Confirm the file exists by running:

ls include/scripts/quality_checks.py

You should see the file path printed back, confirming it was saved.

File not found?

• Make sure you created the file inside include/scripts/ and not in the project root or dags/ folder.
• Confirm the filename is exactly quality_checks.py with an underscore, not a hyphen.

Why isn't my quality_checks.py file showing up?

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before continuing.

ⓧ I'd like to double check the full code

import os
import json
import boto3
import duckdb
from datetime import date
from airflow.exceptions import AirflowException

MINIO_ENDPOINT = os.getenv('MINIO_ENDPOINT', 'http://minio:9000')
MINIO_ACCESS_KEY = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
MINIO_SECRET_KEY = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
DUCKDB_PATH = '/usr/local/airflow/include/warehouse/crypto.duckdb'

def check_bronze_quality():
    today = date.today().isoformat()
    s3 = boto3.client('s3', endpoint_url=MINIO_ENDPOINT, aws_access_key_id=MINIO_ACCESS_KEY, aws_secret_access_key=MINIO_SECRET_KEY)
    obj = s3.get_object(Bucket='bronze', Key=f'assets/{today}/assets.json')
    data = json.loads(obj['Body'].read())
    records = data.get('data', data)
    if len(records) < 10:
        raise AirflowException(f'Bronze quality check failed: only {len(records)} records found')
    for record in records[:5]:
        if not record.get('id') or not record.get('priceUsd'):
            raise AirflowException('Bronze quality check failed: missing required fields')
    print(f'Bronze quality check passed: {len(records)} records')

def check_silver_quality():
    today = date.today().isoformat()
    con = duckdb.connect()
    con.execute('INSTALL httpfs; LOAD httpfs;')
    con.execute(f"SET s3_endpoint='{MINIO_ENDPOINT.replace('http://', '')}'; SET s3_access_key_id='{MINIO_ACCESS_KEY}'; SET s3_secret_access_key='{MINIO_SECRET_KEY}'; SET s3_use_ssl=false; SET s3_url_style='path';")
    count = con.execute(f"SELECT COUNT(*) FROM read_parquet('s3://silver/assets/{today}/*.parquet')").fetchone()[0]
    if count < 10:
        raise AirflowException(f'Silver quality check failed: only {count} rows')
    nulls = con.execute(f"SELECT COUNT(*) FROM read_parquet('s3://silver/assets/{today}/*.parquet') WHERE priceUsd IS NULL OR priceUsd <= 0").fetchone()[0]
    if nulls > 0:
        raise AirflowException(f'Silver quality check failed: {nulls} rows with invalid priceUsd')
    print(f'Silver quality check passed: {count} rows')
    con.close()

def check_gold_quality():
    con = duckdb.connect(DUCKDB_PATH)
    for table in ['gold_asset_rankings', 'gold_market_summary', 'gold_price_aggregations']:
        count = con.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0]
        if count == 0:
            raise AirflowException(f'Gold quality check failed: {table} is empty')
    print('Gold quality check passed')
    con.close()

Add quality check tasks to the DAG

With the validation functions written, you need to wire them into your Airflow DAG as tasks that run between each pipeline stage.

• Open dags/crypto_pipeline.py in your editor.
• Find this line near the top of the file:

from load import load_gold_tables

• Add this import directly below it:

from quality_checks import check_bronze_quality, check_silver_quality, check_gold_quality

• Now find the existing task definitions and dependency chain inside the with DAG(...) block:

    extract_task = PythonOperator(task_id='extract_assets', python_callable=extract_assets)
    transform_task = PythonOperator(task_id='transform_assets', python_callable=transform_assets)
    load_task = PythonOperator(task_id='load_gold_tables', python_callable=load_gold_tables)
    extract_task >> transform_task >> load_task

• Replace that entire block with the following, which adds three quality check tasks and a new dependency chain:

    extract_task = PythonOperator(task_id='extract_assets', python_callable=extract_assets)
    quality_check_bronze = PythonOperator(task_id='quality_check_bronze', python_callable=check_bronze_quality)
    transform_task = PythonOperator(task_id='transform_assets', python_callable=transform_assets)
    quality_check_silver = PythonOperator(task_id='quality_check_silver', python_callable=check_silver_quality)
    load_task = PythonOperator(task_id='load_gold_tables', python_callable=load_gold_tables)
    quality_check_gold = PythonOperator(task_id='quality_check_gold', python_callable=check_gold_quality)
    extract_task >> quality_check_bronze >> transform_task >> quality_check_silver >> load_task >> quality_check_gold

What does this dependency chain do?

The >> operator defines execution order. Your new chain ensures a quality check runs after each stage completes.

• extract_task runs first, then quality_check_bronze validates the raw data.
• Only if bronze passes does transform_task run, followed by quality_check_silver.
• Finally load_task writes to DuckDB, and quality_check_gold confirms the warehouse is healthy.

• Save crypto_pipeline.py.

Seeing an import error in the Airflow UI?

• Make sure the import line says from quality_checks import and not from include.scripts.quality_checks import. The sys.path.insert at the top of the file already points to the include/scripts/ directory.
• Confirm quality_checks.py is in include/scripts/ alongside extract.py, transform.py, and load.py.

Help me fix my DAG import error for quality_checks

✔️ Awesome, I've got everything!

Great. Make sure you have saved the DAG file before moving on.

ⓧ I'd like to double check the full code

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets
from transform import transform_assets
from load import load_gold_tables
from quality_checks import check_bronze_quality, check_silver_quality, check_gold_quality

with DAG(
    dag_id='crypto_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False
) as dag:
    extract_task = PythonOperator(task_id='extract_assets', python_callable=extract_assets)
    quality_check_bronze = PythonOperator(task_id='quality_check_bronze', python_callable=check_bronze_quality)
    transform_task = PythonOperator(task_id='transform_assets', python_callable=transform_assets)
    quality_check_silver = PythonOperator(task_id='quality_check_silver', python_callable=check_silver_quality)
    load_task = PythonOperator(task_id='load_gold_tables', python_callable=load_gold_tables)
    quality_check_gold = PythonOperator(task_id='quality_check_gold', python_callable=check_gold_quality)
    extract_task >> quality_check_bronze >> transform_task >> quality_check_silver >> load_task >> quality_check_gold

Verify the pipeline fails on bad data

The key behaviour you configured is that AirflowException causes Airflow to mark a task as failed. When a quality check task fails, all downstream tasks are skipped automatically.

• Return to the Airflow UI at http://localhost:8080.
• Click the crypto_pipeline DAG to open its detail page.
• Click the **Graph** tab to see the DAG structure.

You should see six nodes in a linear chain: extract_assets then quality_check_bronze then transform_assets then quality_check_silver then load_gold_tables then quality_check_gold.

• Click the **Trigger DAG** button (play icon) to run the pipeline manually.
• Watch the tasks execute in sequence. Each quality check should turn green (success) if your data from previous steps is still in MinIO.

What happens if a check fails?

If any quality check raises AirflowException, that task turns red (failed) in the Graph view. All tasks downstream of the failure are automatically skipped (shown in pink). This prevents bad data from reaching your dashboard.

Quality check task failed unexpectedly?

• Click the failed task node in the Graph view, then click **Log** to see the exact error message.
• If you see a message about fewer than 10 records, your bronze data may be stale. Re-trigger just the extract task first to pull fresh data.
• If you see a connection error to MinIO, make sure all Docker containers are running with astro dev start.

My quality check task is failing, help me debug it

Your pipeline now catches data problems at every layer before they reach your dashboard. Next up, you will connect Metabase to your DuckDB warehouse and build visualizations from the validated data.

Visualize Data with Metabase

Your pipeline is producing clean, structured gold tables in DuckDB. The data is there, but right now the only way to explore it is by writing SQL queries manually.

In this step, you'll connect Metabase to your DuckDB warehouse and build a dashboard that visualizes crypto market trends. This gives anyone on your team a self-serve way to explore the data without writing code.

In this step, get ready to:

• Connect Metabase to the DuckDB warehouse file.
• Build a dashboard with three visualization panels.
• Verify the dashboard updates automatically when the pipeline runs.

Mount the DuckDB warehouse into Metabase

Metabase runs inside a Docker container, so it cannot see files on your host machine unless you explicitly mount them. You need to bind-mount the include/warehouse/ directory into the Metabase container so it can access crypto.duckdb.

• In your code editor, open docker-compose.override.yml in the crypto-pipeline folder.
• Find the metabase service's volumes section. It currently looks like this:

    volumes:
      - metabase-data:/metabase-data

• Add a second volume mount below the existing one, so the section now looks like this:

    volumes:
      - metabase-data:/metabase-data
      - ./include/warehouse:/metabase-data

What does this mount do?

The ./include/warehouse:/metabase-data line maps your local include/warehouse/ directory (which contains crypto.duckdb) into the container at /metabase-data.

When you tell Metabase the database path is /metabase-data/crypto.duckdb, it will read directly from the same file that your Airflow pipeline writes to.

• Save docker-compose.override.yml.
• Restart the stack so the new volume mount takes effect by running:

astro dev stop && astro dev start

• Wait for all services to start. Once the output confirms containers are running, open your browser to http://localhost:3000.

Metabase not loading?

Metabase can take 1-2 minutes to fully start up on first launch. If the page is blank or shows a loading spinner, wait and refresh.

If the container fails to start, run docker logs on the Metabase container to check for port conflicts or volume mount errors.

Help me troubleshoot why Metabase is not starting after I updated docker-compose.override.yml with a new volume mount.

Configure the DuckDB database connection

If this is your first time opening Metabase, you will see a setup wizard. Complete the initial account creation (name, email, password) and skip the default database setup step by clicking I'll add my data later. You can add DuckDB manually in the next step.

• Click the gear icon in the top-right corner of Metabase.
• Select Admin settings.
• Click Databases in the top navigation bar.
• Click Add a database.
• In the Database type dropdown, select DuckDB.
• In the Display name field, enter Crypto Warehouse.
• In the database file path field, enter /metabase-data/crypto.duckdb.
• Click Save changes.

Metabase will connect and begin syncing the database schema. You should see a green success indicator confirming the connection.

Why does DuckDB work as an embedded database here?

The DuckDB driver for Metabase runs DuckDB embedded (in-process) rather than connecting to a separate database server. It reads the .duckdb file directly from the path you provide.

DuckDB only allows one writer at a time. Your Airflow pipeline writes to the file, and Metabase reads from it. Because Metabase only performs read queries for dashboards, there is no write conflict.

DuckDB not appearing in the database type dropdown?

The DuckDB driver JAR must be in the Metabase plugins directory. Verify that the custom Metabase image built successfully by checking that include/metabase/Dockerfile includes the duckdb.metabase-driver.jar download step.

If the driver still does not appear, rebuild the image with astro dev stop followed by astro dev start to force a fresh Docker build.

Help me figure out why DuckDB is not showing in Metabase's database type dropdown.

Build the Crypto Market Overview dashboard

With the database connected, you can build a dashboard that visualizes the gold layer tables. You will create three panels covering market cap rankings, volume leaders, and price distribution.

• Click the + New button in the top-right corner of Metabase.
• Select Dashboard.
• Name the dashboard Crypto Market Overview and click Create.

You are now in the dashboard editor. Add your first panel.

• Click the + icon (or Add a question) to add a new card.
• Select Simple question (or SQL query if you prefer writing SQL directly).
• Choose the Crypto Warehouse database and select the gold_asset_rankings table.
• For the visualization type, choose Bar chart.
• Set the X-axis to symbol and the Y-axis to marketCapUsd.
• Add a filter or limit to show the top 10 rows by rank (ascending).
• Save the card with the title Top Assets by Market Cap.

Now add the second panel for volume leaders.

• Click the + icon again to add another card.
• Select the gold_price_aggregations table from the Crypto Warehouse database.
• For the visualization type, choose Table.
• Sort by volumeUsd24Hr descending and limit to 10 rows.
• Save the card with the title 24h Volume Leaders.

Finally, add the third panel for price distribution.

• Click the + icon to add a third card.
• Select the gold_asset_rankings table from the Crypto Warehouse database.
• For the visualization type, choose Scatter or Table (whichever suits your data best).
• If using scatter, set the X-axis to marketCapUsd and the Y-axis to priceUsd.
• Save the card with the title Price Distribution.
• Click Save on the dashboard to finalize all three panels.

What tables power each panel?

• Top Assets by Market Cap uses gold_asset_rankings which stores the top assets sorted by rank with their market cap.
• 24h Volume Leaders uses gold_price_aggregations which is pre-sorted by volumeUsd24Hr descending.
• Price Distribution uses gold_asset_rankings to show the relationship between market cap and price across all assets.

Tables not showing in Metabase?

If you don't see the gold tables, Metabase may not have finished syncing. Go to Admin settings > Databases > Crypto Warehouse and click Sync database schema.

Also confirm that the pipeline has run at least once so that the gold tables exist in crypto.duckdb.

Help me figure out why Metabase cannot find my DuckDB gold tables.

Verify the dashboard updates when the pipeline runs

The final check is confirming that when Airflow runs the pipeline again and refreshes the gold tables, your Metabase dashboard reflects the new data automatically.

• Open the Airflow UI at http://localhost:8080.
• Find the crypto_pipeline DAG and trigger a manual run by clicking the play button.
• Wait for all tasks to complete successfully (all green).
• Return to your Metabase dashboard at http://localhost:3000.
• Refresh the dashboard page. The panels should reflect the latest data from DuckDB.

How does the dashboard stay up to date?

Metabase queries DuckDB directly every time you load a dashboard. There is no separate caching layer. When the pipeline writes new data to crypto.duckdb, the next dashboard load reads the fresh tables.

Metabase also syncs the database schema hourly by default. If you add new tables in DuckDB, they will appear in Metabase within an hour (or immediately if you trigger a manual sync).

✔️ Awesome, I've got everything!

Great. Double check you've saved your docker-compose.override.yml and that all three panels display data on your dashboard.

ⓧ I'd like to double check the full code

Here is the full docker-compose.override.yml with the Metabase warehouse mount added:

version: '3'
services:
  minio:
    image: minio/minio
    ports:
      - '9000:9000'
      - '9001:9001'
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ':9001'
    volumes:
      - minio-data:/data
  spark-master:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=master
    ports:
      - '7077:7077'
      - '8081:8080'
  spark-worker:
    image: bitnami/spark:latest
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
    depends_on:
      - spark-master
  metabase:
    image: metabase-duckdb:latest
    build:
      context: ./include/metabase
    ports:
      - '3000:3000'
    volumes:
      - metabase-data:/metabase-data
      - ./include/warehouse:/metabase-data
volumes:
  minio-data:
  metabase-data:

Your pipeline now has a visual layer. Anyone can open the Metabase dashboard and explore crypto market data without writing SQL. Next up, you'll add automated tests to make sure the pipeline stays reliable as you iterate.

Write Unit and Integration Tests

Your pipeline extracts, transforms, loads, and validates data. But how do you know the individual pieces still work correctly after you make changes?

Automated tests catch regressions before they reach production. In this step, you will write unit tests that verify your extraction and transformation logic in isolation, plus an integration test that validates the full DuckDB loading flow.

In this step, get ready to:

• Write unit tests for extraction functions (API parsing, error handling).
• Write unit tests for transformation logic (schema validation, business rules).
• Write an integration test validating the full pipeline flow with in-memory DuckDB.

Write unit tests for extraction and transformation

Unit tests verify individual functions in isolation. You will mock external dependencies (the API, MinIO) so the tests run fast without network access.

The Astro CLI already created a tests directory when you initialized the project. You will add your test files there.

• First, add the testing dependencies to requirements.txt. Open requirements.txt in your editor.
• Find the existing four lines (boto3, requests, pyspark, duckdb):

boto3
requests
pyspark
duckdb

• Add pytest and pytest-mock at the bottom, so the file now looks like this:

boto3
requests
pyspark
duckdb
pytest
pytest-mock

• Save requirements.txt.

Why pytest-mock?

The pytest-mock package provides a convenient mocker fixture that wraps Python's built-in unittest.mock. It makes patching external services (like the CoinCap API or MinIO) cleaner in your tests.

Now create the extraction test file.

• In the tests directory of your Astro project, create a new file called test_extract.py.
• Paste the following code into tests/test_extract.py:

import pytest
from unittest.mock import patch, MagicMock
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets

def test_extract_assets_success():
    # Mock a successful API response with one Bitcoin record
    mock_response = MagicMock()
    mock_response.json.return_value = {'data': [{'id': 'bitcoin', 'priceUsd': '50000', 'rank': '1', 'symbol': 'BTC', 'name': 'Bitcoin', 'marketCapUsd': '1000000', 'volumeUsd24Hr': '50000', 'changePercent24Hr': '1.5'}]}
    mock_response.raise_for_status = MagicMock()
    with patch('requests.get', return_value=mock_response), patch('boto3.client') as mock_s3:
        mock_s3.return_value.put_object = MagicMock()
        result = extract_assets()
        # Verify the function returns a valid S3 key
        assert result is not None
        assert 'assets.json' in result

def test_extract_assets_api_error():
    # Verify that an API error propagates as an exception
    with patch('requests.get') as mock_get:
        mock_get.return_value.raise_for_status.side_effect = Exception('API Error')
        with pytest.raises(Exception):
            extract_assets()

What does this code do?

• test_extract_assets_success patches requests.get and boto3.client so no real network calls happen. It verifies that extract_assets() returns a key containing assets.json.
• test_extract_assets_api_error simulates a failed API call by making raise_for_status throw an exception. It confirms the function does not silently swallow errors.

• Save tests/test_extract.py.

Next, create the transformation test file.

• In the tests directory, create a new file called test_transform.py.
• Paste the following code into tests/test_transform.py:

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType

@pytest.fixture(scope='module')
def spark():
    return SparkSession.builder.master('local').appName('test').getOrCreate()

def test_transform_schema(spark):
    # Create a DataFrame matching the expected silver schema
    data = [('bitcoin', 1, 'BTC', 'Bitcoin', 50000.0, 1000000.0, 50000.0, 1.5)]
    schema = StructType([StructField('id', StringType()), StructField('rank', LongType()), StructField('symbol', StringType()), StructField('name', StringType()), StructField('priceUsd', DoubleType()), StructField('marketCapUsd', DoubleType()), StructField('volumeUsd24Hr', DoubleType()), StructField('changePercent24Hr', DoubleType())])
    df = spark.createDataFrame(data, schema)
    # Verify required columns exist
    assert 'priceUsd' in df.columns
    assert 'id' in df.columns

def test_transform_no_negative_prices(spark):
    # Include one valid record and one with a negative price
    data = [('bitcoin', 1, 'BTC', 'Bitcoin', 50000.0, 1000000.0, 50000.0, 1.5), ('bad', 2, 'BAD', 'Bad', -1.0, 0.0, 0.0, 0.0)]
    schema = StructType([StructField('id', StringType()), StructField('rank', LongType()), StructField('symbol', StringType()), StructField('name', StringType()), StructField('priceUsd', DoubleType()), StructField('marketCapUsd', DoubleType()), StructField('volumeUsd24Hr', DoubleType()), StructField('changePercent24Hr', DoubleType())])
    df = spark.createDataFrame(data, schema)
    # Filter out invalid prices (same logic as transform_assets)
    valid_df = df.filter(df.priceUsd > 0)
    assert valid_df.count() == 1

What does this code do?

• The spark fixture creates a local SparkSession shared across all tests in the module. This avoids spinning up a new session per test.
• test_transform_schema validates that the expected columns (priceUsd, id) are present in the DataFrame schema.
• test_transform_no_negative_prices confirms that filtering on priceUsd > 0 removes invalid records. This mirrors the dropna logic in transform_assets().

• Save tests/test_transform.py.

Seeing import errors?

Make sure pyspark is listed in your requirements.txt. The Astro CLI installs these dependencies inside the test container when you run astro dev pytest.

help me debug PySpark import errors in my Astro project tests

Write an integration test for the full pipeline flow

Integration tests verify that multiple components work together. This test uses an in-memory DuckDB instance to validate the loading logic without needing MinIO or Spark running.

• In the tests directory, create a new file called test_integration.py.
• Paste the following code into tests/test_integration.py:

import pytest
import duckdb

def test_full_pipeline_flow():
    # Simulate the gold layer by creating a table in memory
    con = duckdb.connect(':memory:')
    con.execute("CREATE TABLE test_assets AS SELECT 'bitcoin' as id, 1 as rank, 'BTC' as symbol, 'Bitcoin' as name, 50000.0 as priceUsd, 1000000000.0 as marketCapUsd")
    # Verify the table has exactly one row
    result = con.execute('SELECT COUNT(*) FROM test_assets').fetchone()[0]
    assert result == 1
    # Verify the data is queryable and correct
    row = con.execute("SELECT priceUsd FROM test_assets WHERE id = 'bitcoin'").fetchone()
    assert row[0] == 50000.0

What does this test validate?

This test creates a DuckDB table in memory with the same column structure as your gold tables. It confirms that DuckDB can store and query asset data correctly.

By using :memory: instead of a file path, the test runs instantly without touching your real crypto.duckdb warehouse.

• Save tests/test_integration.py.

Run all tests with the Astro CLI

The Astro CLI has a built-in pytest runner that executes tests inside a Docker container with all your project dependencies available.

• Run your entire test suite by running this command:

astro dev pytest

What does this command do?

The astro dev pytest command creates a local Python environment using your project's Dockerfile, requirements.txt, and .env file. It then runs all test files in the tests directory and displays results in your terminal.

• Verify the output shows all tests passing. You should see a summary line with all tests marked as passed.

Tests failing or container not starting?

• Make sure your Astro project's Docker environment is running. If you stopped it, restart with astro dev start before running tests.
• Check that pytest and pytest-mock are both listed in requirements.txt. Missing dependencies cause import failures inside the test container.
• If PySpark tests timeout, they may need more memory. The spark fixture uses local mode inside the container, which is lightweight but still needs Docker to have sufficient resources allocated.

help me debug failing tests in my Astro CLI project

✔️ Awesome, I've got everything!

Great. Double check you have saved all three test files and requirements.txt.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt:

boto3
requests
pyspark
duckdb
pytest
pytest-mock

Here is the complete tests/test_extract.py:

import pytest
from unittest.mock import patch, MagicMock
import sys
sys.path.insert(0, '/usr/local/airflow/include/scripts')
from extract import extract_assets

def test_extract_assets_success():
    mock_response = MagicMock()
    mock_response.json.return_value = {'data': [{'id': 'bitcoin', 'priceUsd': '50000', 'rank': '1', 'symbol': 'BTC', 'name': 'Bitcoin', 'marketCapUsd': '1000000', 'volumeUsd24Hr': '50000', 'changePercent24Hr': '1.5'}]}
    mock_response.raise_for_status = MagicMock()
    with patch('requests.get', return_value=mock_response), patch('boto3.client') as mock_s3:
        mock_s3.return_value.put_object = MagicMock()
        result = extract_assets()
        assert result is not None
        assert 'assets.json' in result

def test_extract_assets_api_error():
    with patch('requests.get') as mock_get:
        mock_get.return_value.raise_for_status.side_effect = Exception('API Error')
        with pytest.raises(Exception):
            extract_assets()

Here is the complete tests/test_transform.py:

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType

@pytest.fixture(scope='module')
def spark():
    return SparkSession.builder.master('local').appName('test').getOrCreate()

def test_transform_schema(spark):
    data = [('bitcoin', 1, 'BTC', 'Bitcoin', 50000.0, 1000000.0, 50000.0, 1.5)]
    schema = StructType([StructField('id', StringType()), StructField('rank', LongType()), StructField('symbol', StringType()), StructField('name', StringType()), StructField('priceUsd', DoubleType()), StructField('marketCapUsd', DoubleType()), StructField('volumeUsd24Hr', DoubleType()), StructField('changePercent24Hr', DoubleType())])
    df = spark.createDataFrame(data, schema)
    assert 'priceUsd' in df.columns
    assert 'id' in df.columns

def test_transform_no_negative_prices(spark):
    data = [('bitcoin', 1, 'BTC', 'Bitcoin', 50000.0, 1000000.0, 50000.0, 1.5), ('bad', 2, 'BAD', 'Bad', -1.0, 0.0, 0.0, 0.0)]
    schema = StructType([StructField('id', StringType()), StructField('rank', LongType()), StructField('symbol', StringType()), StructField('name', StringType()), StructField('priceUsd', DoubleType()), StructField('marketCapUsd', DoubleType()), StructField('volumeUsd24Hr', DoubleType()), StructField('changePercent24Hr', DoubleType())])
    df = spark.createDataFrame(data, schema)
    valid_df = df.filter(df.priceUsd > 0)
    assert valid_df.count() == 1

Here is the complete tests/test_integration.py:

import pytest
import duckdb

def test_full_pipeline_flow():
    con = duckdb.connect(':memory:')
    con.execute("CREATE TABLE test_assets AS SELECT 'bitcoin' as id, 1 as rank, 'BTC' as symbol, 'Bitcoin' as name, 50000.0 as priceUsd, 1000000000.0 as marketCapUsd")
    result = con.execute('SELECT COUNT(*) FROM test_assets').fetchone()[0]
    assert result == 1
    row = con.execute("SELECT priceUsd FROM test_assets WHERE id = 'bitcoin'").fetchone()
    assert row[0] == 50000.0

Your pipeline now has a safety net. Next up, you will set up a CI/CD workflow on GitHub Actions so these tests run automatically on every push.

Set Up CI/CD with GitHub Actions

Your pipeline works locally. Tests pass. DAGs parse. But right now, there is no automated safety net protecting your project from broken code reaching the main branch.

In this step, you will create a GitHub Actions workflow that automatically lints your DAGs, runs your tests, and builds your Docker image on every push. This is how production data teams catch problems before they reach production.

In this step, get ready to:

• Create a GitHub repository and push your project code.
• Write a GitHub Actions CI workflow with DAG linting, tests, and Docker build.
• Verify the CI pipeline passes on GitHub.

Create a GitHub repository and push your project

Before you can automate anything, your code needs to live in a remote repository. You will create a .gitignore file first to keep sensitive and generated files out of version control.

• In your terminal, make sure you are inside the crypto-pipeline folder.
• Create a .gitignore file by running this command:

touch .gitignore

• Open .gitignore and paste the following content:

.env
__pycache__/
*.pyc
include/warehouse/*.duckdb
.astro/

What does this file do?

The .gitignore file tells Git which files to exclude from version control.

• .env contains your API key. Never commit secrets to a repository.
• __pycache__/ and *.pyc are Python bytecode files that get regenerated automatically.
• include/warehouse/*.duckdb is your local warehouse binary. It is large and environment-specific.
• .astro/ contains Astro CLI internal files that differ per machine.

• Save .gitignore.
• Confirm the file exists by running:

ls -a

• You should see .gitignore listed in the output alongside your other project files.

Don't see .gitignore in the output?

• Files starting with a dot are hidden by default. The -a flag in ls -a shows hidden files. If you used ls without the flag, try again with ls -a.
• Make sure you ran touch .gitignore inside the crypto-pipeline folder, not a parent directory.

Help me find my .gitignore file.

• Go to GitHub in your browser and create a new repository named crypto-pipeline.
• Leave it empty (no README, no .gitignore, no license).
• Back in your terminal, initialize Git and push your code by running these commands:

git init
git add .
git commit -m "Initial commit: full crypto pipeline"
git branch -M main
git remote add origin https://github.com/[[GITHUB_USERNAME="your-github-username"]]/crypto-pipeline.git
git push -u origin main

What do these commands do?

• git init creates a new Git repository in your project folder.
• git add . stages all files (except those in .gitignore) for commit.
• git branch -M main renames the default branch to main.
• git remote add origin links your local repo to the GitHub remote.
• git push -u origin main uploads your code to GitHub.

• Refresh your GitHub repository page in the browser.
• You should see all your project files (dags/, include/, tests/, Dockerfile, requirements.txt) listed in the repository.

Push was rejected or authentication failed?

• If you see a permission error, make sure you are authenticated with GitHub. You may need to set up a personal access token or SSH key.
• If the push is rejected because the remote has content, delete the repository on GitHub and recreate it as completely empty.

Help me push my Astro project to GitHub.

Write the GitHub Actions CI workflow

A CI/CD workflow defines automated steps that run every time you push code. Your workflow will lint DAGs, run tests, and build the Docker image to catch issues before they reach production.

• Create the workflow directory by running this command:

mkdir -p .github/workflows

• Create and open a new file at .github/workflows/ci.yml.
• Paste the following workflow configuration:

name: Crypto Pipeline CI

on:
  push:
    branches: ['**']
  pull_request:
    branches: [main]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3
      - name: Install Astro CLI
        run: curl -sSL https://install.astronomer.io | sudo bash -s
      - name: Lint DAGs
        run: astro dev parse
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: pip install -r requirements.txt pytest pytest-mock
      - name: Run tests
        run: pytest tests/ -v
      - name: Build Docker image
        run: docker build -t crypto-pipeline:ci .

What does this workflow do?

Each step in the workflow catches a different type of problem:

• Checkout code pulls your latest commit into the runner.
• Install Astro CLI installs the Astro CLI on the Ubuntu runner using the official installer script.
• Lint DAGs runs astro dev parse to catch syntax and import errors in your DAG files.
• Run tests executes your pytest suite to validate extraction, transformation, and integration logic.
• Build Docker image confirms the project builds into a valid container image.

Why CI/CD for a data pipeline?

Production data teams treat pipelines like software. A broken DAG can silently halt data delivery for hours. Automated checks on every push ensure that bad code never makes it to production without being caught first.

The astro dev parse step is especially important. It validates that all your DAGs can render without syntax or import errors.

• Save .github/workflows/ci.yml.

YAML indentation errors?

YAML is whitespace-sensitive. Make sure you use spaces (not tabs) and that each nested level is indented by exactly 2 spaces. Most editors show a wavy underline if YAML is malformed.

Help me fix my GitHub Actions YAML syntax.

Push and verify the CI pipeline passes

Now you will push the workflow file to GitHub and watch the CI pipeline run. This is the moment where automation takes over.

• Commit and push the new workflow file by running these commands:

git add .github/workflows/ci.yml
git commit -m "Add CI/CD workflow with DAG linting and tests"
git push

• Open your GitHub repository in your browser.
• Click the Actions tab at the top of the repository.
• You should see a workflow run in progress or recently completed. Click on it to see the details.
• Wait for all steps to show a green checkmark. This confirms your DAGs parse correctly, your tests pass, and your Docker image builds successfully.

Seeing a red X on the workflow?

• Click on the failed step to see the error log. The most common issues are missing dependencies in requirements.txt or a DAG import error that astro dev parse catches.
• If the Build Docker image step fails, check that your Dockerfile is committed and contains the correct base image.
• If the Lint DAGs step fails, the error message will show which DAG file has a syntax or import issue.

Help me debug my failing GitHub Actions workflow.

✔️ Awesome, I've got everything!

Great. Double check you have saved both .gitignore and .github/workflows/ci.yml and that your CI run is green on GitHub.

ⓧ I'd like to double check the full code

Here is the complete .gitignore file:

.env
__pycache__/
*.pyc
include/warehouse/*.duckdb
.astro/

Here is the complete .github/workflows/ci.yml file:

name: Crypto Pipeline CI

on:
  push:
    branches: ['**']
  pull_request:
    branches: [main]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3
      - name: Install Astro CLI
        run: curl -sSL https://install.astronomer.io | sudo bash -s
      - name: Lint DAGs
        run: astro dev parse
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: pip install -r requirements.txt pytest pytest-mock
      - name: Run tests
        run: pytest tests/ -v
      - name: Build Docker image
        run: docker build -t crypto-pipeline:ci .

Your pipeline is now fully automated. Every push to any branch triggers DAG linting, test execution, and a Docker image build. Next up, you will take on the Secret Mission and add incremental loading with SCD Type 2 tracking.

Secret mission

Add Incremental Loading With SCD Type 2 Tracking

Your pipeline currently does a full reload every time it runs. In production, that wastes API credits and hides historical trends. Upgrade your pipeline with incremental extraction, deduplication, and a Slowly Changing Dimension table that tracks how crypto rankings shift over time.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers managed by the Astro CLI. There are no ongoing cloud costs.

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

Resources you used:

• Docker containers (Airflow, MinIO, Spark master, Spark worker, Metabase).
• Docker volumes (minio-data, metabase-data).
• Local project folder crypto-pipeline on your Desktop.
• GitHub repository crypto-pipeline.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with your pipeline.

• Your Docker containers continue running and the Airflow scheduler triggers DAG runs on schedule.
• Resume anytime by navigating to http://localhost:8080 for Airflow or http://localhost:3000 for Metabase.
• The Astro CLI remains installed on your machine for future Airflow projects.

Pause - I'll come back to this later

Shut down running containers to free up memory and CPU. All your data persists in Docker volumes so you can resume later.

• Open your terminal and navigate to the crypto-pipeline folder on your Desktop.
• Stop all containers by running:

astro dev stop

• When you are ready to resume, start everything back up by running:

astro dev start

• Your MinIO data, DuckDB warehouse, and Metabase configuration all persist across restarts.

Delete - I don't want to use this again

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

• Open your terminal and navigate to the crypto-pipeline folder on your Desktop.
• Kill all Astro containers and remove associated volumes by running:

astro dev kill

• Move up one directory and delete the project folder.

macOS

cd ~/Desktop
rm -rf crypto-pipeline

Windows

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

• Delete the GitHub repository by going to your crypto-pipeline repository on GitHub.
• Click Settings in the top navigation bar.
• Scroll to the bottom of the General page and click Delete this repository.
• Confirm the deletion when prompted.

Nice Work!

Nice Work!

You just built a production-grade data platform from scratch. From raw cryptocurrency data flowing through the CoinCap API to a fully visualized Metabase dashboard, your pipeline orchestrates extraction, transformation, loading, quality checks, testing, and deployment in one unified system.

You've learned how to:

• Orchestrate an end-to-end data pipeline using Airflow via the Astro CLI, with MinIO for S3-compatible object storage, PySpark for distributed transformations, and DuckDB as the analytical warehouse.
• Validate data quality at every stage of the Medallion Architecture (Bronze, Silver, Gold) and visualize insights through a Metabase dashboard connected to DuckDB.
• Automate testing and deployment with pytest unit and integration tests run by a GitHub Actions CI/CD workflow that lints DAGs, runs tests, and builds Docker images on every push.
• Secret Mission: Added incremental loading with deduplication and a Slowly Changing Dimension (SCD Type 2) table tracking cryptocurrency price and rank changes over time.

Ready to quiz yourself?