ELT Pipeline: S3, Databricks, dbt, Airflow

Build an end-to-end ELT pipeline from API to S3 to Databricks with dbt and Airflow.

Introduction

30 Second Summary

Every online store generates a constant stream of data. Orders, products, customer carts. Getting that raw data cleaned up and ready for business decisions is the core challenge of data engineering.

In this project, you will build a complete ELT pipeline that pulls e-commerce data from an API, stores it in AWS S3, loads it into Databricks, and transforms it into analytics-ready reports using dbt. The whole pipeline runs as a single Airflow DAG you can trigger and monitor from a web UI.

What You'll Build

You'll trigger a single button in Airflow and watch your pipeline extract fresh product data from an API, land it as raw JSON in cloud storage, load it into a data warehouse, and serve up a clean revenue-by-category report you can query instantly in Databricks.

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

  • A one-click pipeline in Airflow where you trigger a run and watch extract, load, and transform tasks complete in sequence.
  • Revenue-by-category and top-selling-products reports you can query directly in Databricks the moment the pipeline finishes.
  • Date-partitioned raw JSON files in your S3 bucket that you can browse and reprocess at any time.
  • Secret Mission: Add dbt data quality tests and configure Airflow retry logic with failure alerting for production reliability.

Are there any prerequisites?

You'll need Python 3.12, VS Code, a Databricks Free Edition account, and Docker with Airflow already running. You'll create an AWS account in Step 1.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

This project wires together five tools into a single automated pipeline. Committing to the full picture now helps you see how each step connects to the next.

Set Up AWS and Python Environment

Every ELT pipeline starts with somewhere to land raw data. Your pipeline will use AWS S3 as its data lake, storing raw JSON files before they flow into Databricks for transformation.

In this step, you will create the cloud storage layer and wire up the Python libraries that connect everything together. Without this foundation, no data can flow through the pipeline.

In this step, get ready to:
  • Create an AWS Free account with an S3 bucket for raw data.
  • Set up a Python virtual environment with project dependencies.
  • Configure AWS credentials so your scripts can access S3.
Create your AWS Free account and S3 bucket

AWS provides a Free plan with $200 in credits for 6 months. S3 storage is listed as always free on both the Free and Paid plans, so you will not be charged for this project.

  • Open your browser and go to aws.amazon.com/free.
  • Click Create a Free Account.
  • Choose the Free plan (no credit card charge).
  • Complete the sign-up process with your email and a root password.
  • Once signed in, open the Amazon S3 console at https://console.aws.amazon.com/s3/.
  • In the left navigation pane, choose General purpose buckets.
  • Click Create bucket.
  • For Bucket name, enter a unique name like elt-pipeline-yourname-2026 (lowercase letters, numbers, and hyphens only).
  • Leave all other settings as defaults and scroll to the bottom.
  • Click Create bucket.

You should see a green banner confirming your bucket was created. The bucket appears in your General purpose buckets list.

Why do I need a unique bucket name?

S3 bucket names are globally unique across all AWS accounts. Adding your name and the year guarantees no one else has already taken the name.

Create an IAM user with S3 access

Your Python scripts need credentials to upload files to S3. You will create a dedicated IAM user with only S3 permissions, then generate an access key for it.

  • Open the IAM console at https://console.aws.amazon.com/iam/.
  • In the left navigation pane, choose Users.
  • Click Create user.
  • For User name, enter elt-pipeline-user.
  • Click Next.
  • Under Permission options, choose Attach policies directly.
  • Search for AmazonS3FullAccess and select its checkbox.
  • Click Next, then Create user.

Now generate an access key for this user:

  • Click the user name elt-pipeline-user to open the user details page.
  • Select the Security credentials tab.
  • Under Access keys, click Create access key.
  • For the use case, choose Other and click Next.
  • Click Create access key.
  • Click Download .csv file to save your Access Key ID and Secret Access Key. Store this file somewhere safe.
  • Click Done.

Save your credentials now

This is your only chance to see the Secret Access Key. If you lose it, you will need to delete the access key and create a new one. Keep the downloaded CSV in a safe place outside your project folder.

Set up your Python virtual environment

You need an isolated Python environment so the project's dependencies stay separate from your system Python. You will create a project folder, a virtual environment, and install the libraries your pipeline needs.

  • Open your terminal.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir elt-pipeline
cd elt-pipeline
  • Create a Python virtual environment by running:
python3 -m venv .venv
  • Activate the virtual environment:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

Your terminal prompt now shows (.venv) at the beginning, confirming the virtual environment is active.

  • Install the project dependencies by running:
pip install boto3 requests dbt-databricks

What are these packages?

  • boto3 is the AWS SDK for Python. It lets your scripts upload and download files from S3.
  • requests makes HTTP calls to APIs. You will use it to fetch data from the DummyJSON API.
  • dbt-databricks is the dbt adapter for Databricks. Installing it also installs dbt-core automatically.

The install takes a minute or two. You should see output ending with a line that says packages were successfully installed.

Install failing?

  • Make sure your terminal shows (.venv) at the start of the prompt. If not, re-run the activation command.
  • If you see a permissions error, confirm you are inside the virtual environment and not using your system Python.

Help me troubleshoot my pip install error in this ELT pipeline project.

Configure AWS credentials and verify the connection

Your Python scripts need to find your AWS credentials automatically. The standard approach is a credentials file at ~/.aws/credentials that boto3 reads by default.

  • Create the AWS config directory by running:

macOS

mkdir -p ~/.aws

Windows

mkdir $HOME\.aws -Force
  • Open VS Code in your project folder by running:
code .
  • In VS Code, create a new file at the path ~/.aws/credentials (macOS/Linux) or C:\Users\YourName\.aws\credentials (Windows) with the following content:
[default]
aws_access_key_id = your-access-key-id
aws_secret_access_key = your-secret-access-key
  • Replace your-access-key-id with the Access Key ID from your downloaded CSV file.
  • Replace your-secret-access-key with the Secret Access Key from your downloaded CSV file.
  • Save the file.

Why a credentials file instead of environment variables?

The ~/.aws/credentials file is the standard location that all AWS SDKs check automatically. It keeps your keys out of your project code and works across all Python scripts without extra configuration.

Now verify that boto3 can reach your S3 bucket. Before you run the next command, what do you expect to see if your credentials are configured correctly?

  • Test the connection by running:
python -c "import boto3; s3 = boto3.client('s3'); print(s3.list_buckets())"

You should see a dictionary that includes your bucket name in the output. This proves boto3 is reading your credentials and connecting to AWS successfully.

Finally, run the full verification that confirms all packages are importable:

  • Verify all dependencies are installed by running:
python -c "import boto3; import requests; print('Ready')"

You should see Ready printed in your terminal. Your environment is fully set up.

Not seeing Ready?

  • If you see ModuleNotFoundError, make sure your virtual environment is still active (look for (.venv) in the prompt). Re-run the activation command if needed.
  • If the list_buckets call fails with a credentials error, double-check that your ~/.aws/credentials file has the correct key values from the CSV you downloaded.

Help me debug my AWS credentials or Python import issue.

Your cloud storage and Python environment are ready to go. Next up, you will write a script that pulls real data from an API and lands it in your S3 bucket.

Extract Data and Land in S3

Your AWS account is ready and your Python libraries are installed. Now it is time to build the first real piece of the pipeline: pulling data out of a source system and landing it in your data lake.

The first step of any ELT pipeline is extraction. You will write a Python script that fetches e-commerce data from the DummyJSON API and uploads the raw JSON files to S3 with date-partitioned paths. This gives you a timestamped landing zone you can always trace back to.

In this step, get ready to:
  • Fetch all products and carts from the DummyJSON API using the requests library.
  • Upload the raw JSON responses to S3 under date-partitioned paths using boto3.
  • Confirm both files are visible in your S3 bucket.
Write the extraction logic

The DummyJSON API provides realistic e-commerce data with no authentication required. You will fetch two datasets: products (with prices and categories) and carts (with nested product arrays and quantities).

  • Create a new file called extract.py in your project directory by running:
touch extract.py
  • Open extract.py in VS Code and paste the following extraction code:
import requests
import boto3
import json
from datetime import date

# Your S3 bucket name from Step 1
BUCKET_NAME = "[[BUCKET_NAME=\"your-bucket-name-from-step-1\"]]"
today = date.today().isoformat()

# Extract all products from DummyJSON API
products_response = requests.get("https://dummyjson.com/products?limit=0")
products_data = products_response.json()
print(f"Fetched {len(products_data['products'])} products")

What does this code do?

The limit=0 query parameter tells DummyJSON to return all items at once instead of paginating. The API currently has 194 products.

The today variable captures the current date as a string like 2026-08-09. You will use this to organize files by date in S3.

  • Save extract.py and run it to confirm the API call works:
python extract.py

You should see Fetched 194 products printed in your terminal.

Seeing a connection error?

Make sure your virtual environment is activated (you should see (.venv) in your terminal prompt). If the request times out, check your internet connection and try again.

Help me debug why my extract.py cannot reach the DummyJSON API.

Add carts extraction and S3 upload

Now you will fetch the carts data and add the code that uploads both JSON responses to S3. Each file lands under a date-partitioned path so you can always see when data was extracted.

  • In extract.py, add the following code below the products extraction:
# Extract all carts from DummyJSON API
carts_response = requests.get("https://dummyjson.com/carts?limit=0")
carts_data = carts_response.json()
print(f"Fetched {len(carts_data['carts'])} carts")

# Upload to S3 with date-partitioned paths
s3 = boto3.client("s3")

s3.put_object(
    Bucket=BUCKET_NAME,
    Key=f"raw/products/{today}/products.json",
    Body=json.dumps(products_data)
)
print(f"Uploaded raw/products/{today}/products.json")

s3.put_object(
    Bucket=BUCKET_NAME,
    Key=f"raw/carts/{today}/carts.json",
    Body=json.dumps(carts_data)
)
print(f"Uploaded raw/carts/{today}/carts.json")

print("Extraction complete!")

What does this code do?

The carts endpoint returns all carts, each containing a nested products array with item-level quantities and totals. This nested structure becomes important later when you try to query it directly.

The s3.put_object() calls upload each JSON response as a file. The Key parameter sets the path inside your bucket using today's date, creating a structure like raw/products/2026-08-09/products.json.

Why date-partition raw data?

Date-partitioned paths let you run the same script daily and build up a history of extractions. If something goes wrong with a later run, you can always trace back to the exact data that was loaded on any given day.

  • Save extract.py.
Run the script and verify files in S3

Before you run this, what do you expect to see printed in your terminal? Think about how many print statements are in the script and what each one outputs.

  • Run the full extraction script:
python extract.py

You should see five lines of output: the product count, the cart count, two upload confirmations, and the completion message.

What should I see?

Your terminal output looks similar to this:

Fetched 194 products
Fetched 50 carts
Uploaded raw/products/2026-08-09/products.json
Uploaded raw/carts/2026-08-09/carts.json
Extraction complete!
  • Open the AWS S3 console in your browser.
  • Navigate to your bucket and open the raw/ folder.
  • Confirm you see products/ and carts/ folders, each containing a date folder with a JSON file inside.

Not seeing any files in S3?

Double-check that the BUCKET_NAME value in extract.py matches the exact bucket name you created in Step 1. Bucket names are case-sensitive and globally unique.

Also verify your AWS credentials are configured correctly. If you see an access denied error, your IAM user may not have the correct S3 permissions attached.

Help me debug why my S3 upload is failing.

✔️ Awesome, I've got everything!

Great. Double check you have saved extract.py and confirmed both files are visible in your S3 bucket.

ⓧ I'd like to double check the full code

Your extract.py should contain two sections in order. Scroll up and confirm you have both:

  • The imports, BUCKET_NAME config, today variable, and products extraction with a print statement (first chunk).
  • The carts extraction, S3 client creation, both put_object calls with print statements, and the completion message (second chunk).

Your data lake now has raw e-commerce data sitting in S3. Next up, you will load this data into Databricks and discover firsthand why querying raw nested JSON is painful.

Load Raw Data into Databricks

Your raw data is sitting safely in S3, but a data lake alone cannot answer business questions. To run SQL queries and build analytics, you need the data inside a proper query engine.

In this step, you will load the raw JSON from S3 into Databricks Delta tables. Then you will try to answer a real business question directly from the raw data. The difficulty you experience is intentional and sets up why dbt exists.

In this step, get ready to:
  • Create a Databricks notebook that reads raw JSON from S3 into Delta tables.
  • Attempt to answer a business question from raw nested data and experience the pain firsthand.
Create a notebook and configure S3 access

Databricks notebooks let you run PySpark code interactively against your data warehouse. You will create a notebook and configure it to read from your S3 bucket using your AWS credentials.

  • Open your Databricks Free Edition workspace in your browser.
  • Click + New in the left sidebar.
  • Select Notebook from the menu.
  • Name the notebook load_raw by clicking the title at the top.

What is a Databricks notebook?

A Databricks notebook is a web-based code editor where you write PySpark or SQL in individual cells and run them one at a time. In Free Edition, your notebook automatically connects to serverless compute, so there is no cluster to configure.

The notebook needs AWS credentials to reach your S3 bucket. Paste the following code into the first empty cell to configure S3 access:

# Configure AWS credentials so Spark can read from your S3 bucket
spark.conf.set("fs.s3a.access.key", "your-access-key-id-here")
spark.conf.set("fs.s3a.secret.key", "your-secret-access-key-here")
  • Replace your-access-key-id-here with the Access Key ID you saved in Step 1.
  • Replace your-secret-access-key-here with the Secret Access Key you saved in Step 1.
  • Press Shift+Enter to run the cell and create a new cell below it.

What does this code do?

The fs.s3a.access.key and fs.s3a.secret.key properties tell Spark's S3A filesystem connector how to authenticate with AWS. Once set, any s3a:// path you reference in this notebook session uses these credentials.

Load products from S3 into a Delta table

The DummyJSON API response is a single JSON object (not line-delimited), so Spark needs the multiLine option enabled. The products sit inside a top-level products array that you will explode into individual rows.

  • Paste the following code into the next cell to read and save your products data:
from pyspark.sql.functions import explode, col

# Set the path to your raw products JSON in S3
bucket = "your-bucket-name-here"
date = "2026-08-09"
products_path = f"s3a://{bucket}/raw/products/{date}/products.json"

# Read the JSON file (single object, so multiLine is required)
products_raw = spark.read.option("multiLine", "true").json(products_path)

# Explode the products array into individual rows
products_df = products_raw.select(explode(col("products")).alias("product")).select("product.*")

# Write as a Delta table
products_df.write.mode("overwrite").saveAsTable("raw_products")
  • Replace your-bucket-name-here with your actual S3 bucket name (e.g. elt-pipeline-yourname-2026).
  • Replace the date value with the date you ran extract.py (the folder name visible in your S3 bucket).
  • Press Shift+Enter to run the cell.

What does this code do?

The multiLine option tells Spark the entire file is one JSON object rather than one-record-per-line. The explode function unpacks the products array so each product becomes its own row in the DataFrame.

The saveAsTable call writes the data as a Delta table in your default schema. Delta is the native storage format in Databricks, giving you versioning, ACID transactions, and fast SQL access.

The cell should complete without errors. Now verify the table exists by running a quick query.

  • Paste this into the next cell to preview the products table:
# Verify raw_products loaded correctly
spark.sql("SELECT id, title, price, category FROM raw_products LIMIT 5").display()
  • Press Shift+Enter to run the cell.
  • You should see a table with 5 rows showing product IDs, titles, prices, and categories.

Seeing an access denied or connection error?

Double-check that your Access Key ID and Secret Access Key are pasted correctly in Cell 1 with no extra spaces. Re-run Cell 1 after fixing, then re-run Cell 2.

If you see a path-not-found error, confirm the date folder in your S3 bucket matches the date variable exactly. You can check the correct folder name in the AWS S3 console.

Still stuck? Why can't my Databricks notebook connect to S3?

Load carts and attempt a revenue query

Now load the carts data using the same pattern. Then you will try to answer a business question directly from the raw tables.

  • Paste the following code into the next cell to load carts:
# Read the raw carts JSON from S3
carts_path = f"s3a://{bucket}/raw/carts/{date}/carts.json"

# Read and explode the carts array into individual rows
carts_raw = spark.read.option("multiLine", "true").json(carts_path)
carts_df = carts_raw.select(explode(col("carts")).alias("cart")).select("cart.*")

# Write as a Delta table
carts_df.write.mode("overwrite").saveAsTable("raw_carts")
  • Press Shift+Enter to run the cell.

What does this code do?

This follows the same pattern as products: read the multi-line JSON, explode the top-level carts array into one row per cart, and save as a Delta table called raw_carts.

  • Verify the carts table by pasting this into the next cell:
# Preview raw_carts structure
spark.sql("SELECT * FROM raw_carts LIMIT 3").display()
  • Press Shift+Enter to run the cell.
  • You should see cart rows. Notice the products column contains a nested array of objects rather than simple values.

Both Delta tables are loaded. Now comes the real test.

Imagine your manager asks: "What is the total revenue per product category?" This is a common analytics question. Before you run the next cell, think about what you would need to do to answer this from raw_carts alone. What obstacles do you expect?

  • Paste this query into the next cell to try answering the revenue question from raw data:
-- Attempt: total revenue per product category from raw data
-- Problem 1: products inside each cart are a nested array
-- Problem 2: cart products don't have a category field
-- Problem 3: we need to join with raw_products to get categories

SELECT
  p.category,
  SUM(cp.total) as revenue
FROM raw_carts c
LATERAL VIEW explode(c.products) AS cp
JOIN raw_products p ON cp.id = p.id
GROUP BY p.category
ORDER BY revenue DESC
  • Change the cell language to SQL by clicking the language dropdown at the top-left of the cell and selecting SQL.
  • Press Shift+Enter to run the cell.

This query may fail or produce confusing results. The nested products array inside each cart needs special SQL syntax to unpack. Column names inside the nested struct may not match what you expect. The join logic is fragile because the raw data was never designed for direct analysis.

Why is this so painful?

You just experienced the core problem that dbt solves. Raw data from APIs arrives nested, inconsistently named, and not joinable without gymnastics. Three specific issues make this query hard:

  • Cart products are a nested array. You need LATERAL VIEW explode or similar syntax just to get individual items.
  • Cart product records have no category field. You must join with the products table, but the raw schema makes the join key ambiguous.
  • Every analyst who needs this data would have to re-discover and re-write this complex query from scratch.

In the next step, dbt staging models will flatten and clean this data once. Then the revenue query becomes a simple three-line SQL statement.

Your raw data is now in Databricks, but the pain of querying it is fresh. In the next step, you will use dbt to build clean staging models that solve every one of these problems permanently.

Transform with dbt (Staging and Mart Models)

In the last step, you tried to answer a simple business question from raw data and hit a wall. Nested arrays, missing joins, and messy schemas made a straightforward query painful.

That pain is exactly what dbt solves. dbt lets you write SQL models that flatten, clean, and reshape raw data into analytics-ready tables. In this step, you will build staging models that normalize the raw data and a mart model that answers the revenue-by-category question in a single clean query.

In this step, get ready to:
  • Initialize a dbt project and connect it to Databricks.
  • Create staging models that flatten raw products and unnest cart items.
  • Build a mart model that answers "total revenue per category" cleanly.
Initialize the dbt project and connect to Databricks

Before you can write any SQL models, dbt needs a project structure and a connection to your Databricks warehouse. You will scaffold the project with dbt init and then configure the connection profile.

  • In your terminal, make sure your virtual environment is active, then initialize a new dbt project by running:
dbt init elt_pipeline

What prompts will I see?

dbt init is interactive. It will ask you which database adapter to use. Select the number corresponding to databricks from the list. It may also prompt for connection details. You can skip the profile setup for now (press Enter through the prompts) because you will configure the profile manually in the next chunk.

  • Move into the new project folder by running:
cd elt_pipeline

You now have a standard dbt project directory with a dbt_project.yml file and a models/ folder.

Next, you need to tell dbt how to connect to your Databricks SQL Warehouse. You will need three values from your Databricks workspace:

  • Switch to your Databricks workspace in the browser.
  • Click SQL in the left sidebar, then click SQL Warehouses.
  • Click your SQL Warehouse name, then open the Connection Details tab.
  • Copy the Server hostname and HTTP path values. You will paste them into your profile next.
  • Generate a Personal Access Token (PAT) for authentication.
  • Click your username in the top-right corner, then click Settings.
  • Click Developer in the left panel.
  • Under Access tokens, click Manage.
  • Click Generate new token. Give it a comment like dbt-pipeline and click Generate.
  • Copy the token value immediately. You will not be able to see it again.
  • Create the connection profile by opening profiles.yml in the project root (inside your elt_pipeline folder) and pasting the following:
elt_pipeline:
  target: dev
  outputs:
    dev:
      type: databricks
      host: your-workspace-hostname
      http_path: /sql/1.0/warehouses/your-warehouse-id
      token: your-personal-access-token
      schema: default
      threads: 1

What does this configuration do?

  • type: databricks tells dbt to use the Databricks adapter.
  • host is the Server hostname from Connection Details (e.g. adb-1234567890.12.azuredatabricks.net).
  • http_path is the HTTP path from Connection Details (starts with /sql/1.0/warehouses/).
  • token is the PAT you just generated.
  • schema: default targets the schema where your raw tables already live.
  • Replace your-workspace-hostname, /sql/1.0/warehouses/your-warehouse-id, and your-personal-access-token with the actual values you copied.
  • Save profiles.yml.
  • Verify the connection works by running:
dbt debug

You should see a series of checks ending with "All checks passed!" confirming dbt can reach your Databricks warehouse.

Seeing connection errors?

  • Double-check that host does not include https:// at the start. It should be just the hostname.
  • Confirm the http_path starts with /sql/. A missing leading slash will cause a 404.
  • Make sure your SQL Warehouse is running (not stopped) in the Databricks console.

Still stuck? Help me debug my dbt connection to Databricks

Define sources and create staging models

dbt sources tell dbt about tables that already exist in your warehouse (tables you did not create with dbt). You will define your two raw tables as sources, then build staging models that clean and flatten them.

  • Remove the example models that dbt init created by running:
rm -rf models/example
  • Create the staging and marts directories by running:
mkdir models/staging models/marts
  • Define your raw tables as dbt sources by creating models/sources.yml with the following content:
version: 2

sources:
  - name: default
    tables:
      - name: raw_products
      - name: raw_carts

What is a dbt source?

A source is a table that exists in your warehouse but was not created by dbt. Declaring sources in a YAML file lets you reference them with the {{ source() }} function in your models. This creates a dependency graph so dbt knows where your data comes from.

  • Create the first staging model by adding models/staging/stg_products.sql with this content:
-- Flatten raw products into clean columns
select
    id,
    title,
    price,
    category
from {{ source('default', 'raw_products') }}

What does this model do?

This staging model selects only the four columns you need from the raw products table: id, title, price, and category. It drops dozens of irrelevant fields (thumbnail, images, rating) so downstream models work with a clean, focused table.

  • Confirm the source definition and model compile correctly by running:
dbt compile --select "stg_products"

You should see output ending with "Done" and no errors. This confirms dbt can parse your source reference and generate valid SQL.

Seeing a compilation error?

  • Check that sources.yml is inside models/ (not models/staging/). dbt scans the models directory for YAML files.
  • Verify the source name default and table name raw_products in your YAML match what you used in the SQL model's {{ source() }} call exactly.

Need help? Help me fix my dbt compile error for stg_products

  • Create the second staging model by adding models/staging/stg_carts.sql with this content:
-- Unnest the products array into one row per cart item
select
    c.id as cart_id,
    item.id as product_id,
    item.quantity,
    item.total
from {{ source('default', 'raw_carts') }} as c
lateral view explode(products) as item

What does this model do?

Each row in raw_carts contains a products column that holds an array of structs. Each struct represents one product in that cart with fields like id, quantity, and total.

The LATERAL VIEW EXPLODE clause unpacks that nested array. It produces one output row for each item in the array, turning a single cart row with 5 products into 5 rows, each with a cart_id, product_id, quantity, and total.

  • Confirm both staging models compile successfully by running:
dbt compile

You should see output showing that 2 models compiled without errors.

Build the mart model and run the pipeline

The whole point of staging models is to make downstream analytics trivial. Now you will write a mart model that joins the clean staging tables and answers the revenue-by-category question in a few lines of SQL.

  • Create the mart model by adding models/marts/mart_revenue_by_category.sql with this content:
-- Aggregate total revenue by product category
select
    p.category,
    sum(c.total) as total_revenue
from {{ ref('stg_carts') }} as c
join {{ ref('stg_products') }} as p
    on c.product_id = p.id
group by p.category
order by total_revenue desc

What does this model do?

This mart model joins stg_carts with stg_products on product_id to get the category for each cart item. Then it sums total (the item revenue) by category and sorts highest-revenue categories first.

Compare this to the painful raw query you attempted in Step 3. The staging layer did the hard work of flattening and cleaning, so the mart is just a simple join and aggregate.

Before you run this, think about what you expect to happen. You have 3 model files (2 staging, 1 mart). What do you think dbt run will produce?

  • Execute all your dbt models by running:
dbt run

You should see output showing 3 models completing successfully. dbt creates each model as a table (or view) in your Databricks schema, running them in dependency order: staging models first, then the mart.

Seeing a runtime error on stg_carts?

  • If you see an error about products not being an array type, your raw_carts table may have loaded the JSON as a string column rather than a parsed struct. Return to your Databricks notebook from Step 3 and confirm the table schema shows products as an ARRAY<STRUCT> type.
  • If you see "table not found" for raw_carts, check that schema: default in profiles.yml matches the schema where you created the raw tables in Step 3.

Still stuck? Help me debug my dbt run errors on the staging models

  • Verify the mart results by opening a notebook in your Databricks workspace and running this query:
SELECT * FROM mart_revenue_by_category

You should see clean rows showing each product category and its total revenue, sorted from highest to lowest. This is the same question that was painful to answer from raw data in Step 3, now trivially queryable thanks to dbt.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your model files, then move on.

ⓧ I'd like to double check the full code

Here are the key files you should have in your elt_pipeline project. Confirm each one is present and matches:

  • profiles.yml in the project root (with your actual Databricks values filled in).
  • models/sources.yml defining raw_products and raw_carts as sources.
  • models/staging/stg_products.sql selecting id, title, price, category from the raw_products source.
  • models/staging/stg_carts.sql using LATERAL VIEW EXPLODE to unnest the products array.
  • models/marts/mart_revenue_by_category.sql joining stg_carts with stg_products and aggregating by category.

Your raw data is now transformed into clean, analytics-ready tables. Next up, you will wire the entire pipeline together with Airflow so it runs automatically as a single orchestrated workflow.

Orchestrate with Airflow

You now have all three pieces of your pipeline working independently. You can extract data, load it into Databricks, and transform it with dbt.

The problem is that running scripts manually is not a pipeline. If you forget to run extract.py before the load step, or skip dbt run after loading, your data goes stale. Apache Airflow solves this by turning your separate steps into a single DAG (Directed Acyclic Graph) that runs in sequence, handles failures, and gives you a visual UI to monitor everything.

In this step, get ready to:
  • Create an Airflow DAG file with three sequential tasks.
  • Make your Python dependencies available inside the Airflow Docker container.
  • Trigger the full pipeline from the Airflow UI and watch all tasks turn green.
Create the DAG file

An Airflow DAG defines which tasks run and in what order. You will create a single DAG with three tasks that mirror the three pipeline stages you already built.

  • In VS Code, create a new file called elt_pipeline_dag.py inside your project's dags/ folder (create the folder if it does not exist yet).
  • Add the imports and DAG definition by pasting this code:
from datetime import datetime
import subprocess
import json
import boto3
import requests

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

# Define the DAG and its schedule
with DAG(
    dag_id="elt_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
) as dag:
    pass

What does this code do?

This creates an Airflow DAG named elt_pipeline. The schedule=None means it only runs when you trigger it manually. catchup=False prevents Airflow from running the DAG for every missed day since the start date.

In Airflow 3.x, you import DAG from airflow.sdk and operators from airflow.providers.standard. The pass is a placeholder you will replace with tasks next.

  • Replace pass inside the DAG context manager with the three task functions and their PythonOperator definitions:
    # Task 1: Extract data from API and upload to S3
    def extract_and_land_fn():
        from datetime import date
        today = date.today().isoformat()
        bucket = "elt-pipeline-yourname-2026"

        # Fetch products and carts from DummyJSON
        products = requests.get("https://dummyjson.com/products?limit=0").json()
        carts = requests.get("https://dummyjson.com/carts?limit=0").json()

        # Upload raw JSON to S3 with date partitioning
        s3 = boto3.client("s3")
        s3.put_object(Bucket=bucket, Key=f"raw/products/{today}/products.json", Body=json.dumps(products))
        s3.put_object(Bucket=bucket, Key=f"raw/carts/{today}/carts.json", Body=json.dumps(carts))

What does extract_and_land_fn do?

This function replicates the logic from your extract.py script. It fetches all products and carts from the DummyJSON API, then uploads the raw JSON to your S3 bucket under date-partitioned paths.

  • Below the extract function, add the load function:
    # Task 2: Load raw data from S3 into Databricks Delta tables
    def load_to_databricks_fn():
        # Run the Databricks load notebook via databricks-sql-connector
        # or replicate the PySpark logic from your notebook here
        import subprocess
        subprocess.run(
            ["python", "/opt/airflow/scripts/load_to_databricks.py"],
            check=True
        )

What does load_to_databricks_fn do?

This function calls your load logic as a subprocess. You will place a load_to_databricks.py script (the Python equivalent of your Databricks notebook) in the container's scripts folder. It reads the raw JSON from S3 and writes it into your Databricks Delta tables.

  • Below the load function, add the dbt transform function and all three PythonOperator task definitions:
    # Task 3: Run dbt transformations
    def dbt_transform_fn():
        subprocess.run(
            ["dbt", "run", "--project-dir", "/opt/airflow/dbt/elt_pipeline"],
            check=True
        )

    # Define PythonOperator tasks
    extract_and_land = PythonOperator(
        task_id="extract_and_land",
        python_callable=extract_and_land_fn,
    )

    load_to_databricks = PythonOperator(
        task_id="load_to_databricks",
        python_callable=load_to_databricks_fn,
    )

    dbt_transform = PythonOperator(
        task_id="dbt_transform",
        python_callable=dbt_transform_fn,
    )

    # Set the dependency chain: extract >> load >> transform
    extract_and_land >> load_to_databricks >> dbt_transform

What does the dependency chain do?

The >> operator tells Airflow to run tasks in sequence. extract_and_land must complete before load_to_databricks starts, and load_to_databricks must complete before dbt_transform starts. If any task fails, the downstream tasks do not run.

  • Save dags/elt_pipeline_dag.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file.

ⓧ I'd like to double check the full code

Your complete dags/elt_pipeline_dag.py should contain these blocks in order:

  • Imports (datetime, subprocess, json, boto3, requests, airflow.sdk DAG, PythonOperator).
  • The with DAG(...) context manager with dag_id="elt_pipeline", schedule=None, catchup=False.
  • Three function definitions: extract_and_land_fn, load_to_databricks_fn, dbt_transform_fn.
  • Three PythonOperator definitions with task IDs: extract_and_land, load_to_databricks, dbt_transform.
  • The dependency chain: extract_and_land >> load_to_databricks >> dbt_transform.
Make dependencies available inside the Airflow container

Your DAG uses boto3, requests, and dbt-databricks. These are installed in your local virtual environment, but the Airflow Docker container does not have them yet.

  • Create a requirements.txt file in the same directory as your docker-compose.yaml by running:
echo "boto3
requests
dbt-databricks" > requirements.txt
  • Create a Dockerfile in the same directory with this content:
FROM apache/airflow:3.3.0
ADD requirements.txt .
RUN pip install apache-airflow==${AIRFLOW_VERSION} -r requirements.txt

Why build a custom image?

The official Airflow Docker image does not include your project's Python packages. By extending the image with a Dockerfile and requirements.txt, you bake the dependencies into the image once. This is faster and more reliable than installing packages every time a container starts.

  • In your docker-compose.yaml, comment out the image: line and enable build: . so Docker uses your custom image.
  • Rebuild the Airflow image and restart your containers by running:
docker compose build
docker compose up -d
  • Wait for the containers to start (about 30 seconds), then confirm they are running:
docker compose ps

You should see your Airflow webserver, scheduler, and worker containers listed with a running or healthy status.

Containers not starting?

  • Check docker compose logs for error messages. A common issue is a typo in requirements.txt or a missing Dockerfile.
  • Make sure Docker Desktop is running and has enough memory allocated (at least 4 GB).

Help me debug my Airflow Docker containers not starting

Trigger the DAG and verify the pipeline

Everything is wired up. Time to see if the full pipeline runs end-to-end from a single trigger.

  • Open your browser and go to localhost:8080.
  • Log in with the default credentials (username airflow, password airflow).
  • Find the elt_pipeline DAG in the list.
  • Toggle the switch next to the DAG name to unpause it.

Before you trigger the run, what do you expect to happen? Do you think all three tasks will turn green, or will one fail?

  • Click the play button (triangle icon) on the right side of the DAG row to trigger a manual run.
  • Click on the DAG name to open the graph view.
  • Watch each task execute in sequence: extract_and_land runs first, then load_to_databricks, then dbt_transform.

All three tasks should turn green, indicating successful completion.

Seeing a red (failed) task?

  • Click the failed task in the graph view, then click Log to see the error message.
  • If you see an import error for boto3 or requests, your custom image build did not include the requirements. Run docker compose build --no-cache and then docker compose up -d again.
  • If the dbt_transform task fails, check that the --project-dir path matches where your dbt project is mounted inside the container.

Help me debug a failed task in my Airflow DAG

  • Confirm fresh data landed by switching to your Databricks workspace and running this query in a notebook:
SELECT * FROM mart_revenue_by_category

You should see clean revenue-per-category rows reflecting the data your pipeline just processed end-to-end.

Your pipeline is now fully orchestrated. You can trigger it any time from the Airflow UI and watch all three stages execute in sequence without touching a single script manually. In the Secret Mission, you will add data quality checks and failure handling to make this pipeline production-ready.

Secret mission

Add dbt Tests and Airflow Retry/Alerting

Your pipeline works, but what happens when bad data sneaks in or a task fails at 3 AM? In this mission, you will add dbt tests to catch data quality issues and configure Airflow to automatically retry failed tasks and log alerts. This is the difference between a toy pipeline and a production one.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses AWS Free plan credits for S3 (always free tier) and Databricks Free Edition (no cost), so there are no ongoing charges while resources sit idle.

Resources you used:

  • AWS S3 bucket with raw JSON data files.
  • IAM user with access keys configured at ~/.aws/credentials.
  • Databricks Free Edition workspace with Delta tables and a notebook.
  • Python virtual environment (.venv) with installed packages.
  • Local project files: extract.py, elt_pipeline/ dbt project, and dags/ folder.
  • Docker Airflow containers.

Keep everything running

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

  • Your S3 bucket stores minimal data well within the AWS Free plan's always-free S3 tier.
  • Databricks Free Edition has no charges.
  • Docker containers use local resources only. They will consume memory while running.
  • Your Python virtual environment and local files remain on disk.

Pause - I'll come back to this later

Shut down running processes to free up local memory while keeping all your data and files intact.

  • Stop your Docker Airflow containers by running:
docker-compose down
  • Your S3 data, Databricks tables, and local project files all persist. You can restart Airflow later with docker-compose up.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below.

  • Delete your S3 bucket and its contents by running:
aws s3 rb s3://elt-pipeline-yourname-2026 --force
  • Delete the IAM user you created for this project. Open the IAM console at https://console.aws.amazon.com/iam/, choose Users in the left navigation, select your project IAM user, then choose Delete. Confirm by typing the username.
  • Remove your local AWS credentials file by running:
rm ~/.aws/credentials
  • Drop the Databricks Delta tables. In a Databricks notebook, run these SQL statements:
DROP TABLE IF EXISTS raw_products;
DROP TABLE IF EXISTS raw_carts;
DROP TABLE IF EXISTS stg_products;
DROP TABLE IF EXISTS stg_carts;
DROP TABLE IF EXISTS mart_revenue_by_category;
  • Delete the Databricks notebook by right-clicking it in the workspace sidebar and selecting Move to Trash.
  • Remove Docker Airflow containers and their volumes by running:
docker-compose down -v
  • Delete your local project files. On macOS or Linux, run:
rm -rf .venv extract.py elt_pipeline dags
  • On Windows, run:
Remove-Item -Recurse -Force .venv, extract.py, elt_pipeline, dags

Nice Work!

Nice Work!

You just built a production-pattern ELT pipeline from scratch. Data flows from an API through cloud storage into a warehouse, gets transformed into analytics-ready models, and the whole thing runs as a single orchestrated workflow you can trigger with one click.

Here's what you accomplished:

  • Built a complete ELT pipeline that extracts e-commerce data from the DummyJSON API, lands raw JSON in AWS S3 with date-partitioned paths, and loads it into Databricks Delta tables.
  • Transformed messy nested data into clean, queryable models using dbt with the staging/mart pattern. Revenue-by-category reports that were painful to write against raw JSON became a single clean query.
  • Orchestrated the full extract-load-transform pipeline as an Airflow DAG with sequential task dependencies you can trigger and monitor from a web UI.
  • (Secret Mission) Added dbt tests for data quality validation and configured Airflow retry logic with failure callbacks to catch bad data and handle task failures automatically.

Ready to quiz yourself?