ETL Pipeline: DynamoDB to Snowflake

Extract NoSQL data from DynamoDB, transform with PySpark, and model with dbt.

Introduction

30 Second Summary

Every business collects data, but raw data rarely lives in the format analysts need. Orders might sit as nested documents in a NoSQL database, while the analytics team needs flat, queryable tables with quality guarantees.

In this project, you will build an end-to-end ETL pipeline that moves e-commerce order data from AWS DynamoDB all the way to analytics-ready models in Snowflake. Along the way, you will use PySpark to flatten nested data and dbt to add data quality tests.

What You'll Build

You will run a Python script that pulls nested order documents from DynamoDB through PySpark transformations into Snowflake, then use dbt to build tested analytics models you can query with SQL to answer real business questions.

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

  • A working extraction and transformation pipeline that pulls nested order data from DynamoDB, flattens it with PySpark, and loads clean tabular rows into Snowflake.
  • Analytical SQL queries that run against your loaded data to find top customers, revenue by product, and order trends.
  • A dbt project with staging models, mart models, and data quality tests that pass when you run dbt test.
  • Secret Mission: Add incremental materialization and source freshness monitoring to make your pipeline production-ready.

Are there any prerequisites?

You will need Python 3.10, Java/JDK (required by PySpark), an existing AWS account, and a Snowflake account. Familiarity with Python basics and SQL fundamentals is assumed.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters. This is the moment you commit to the project goal before any hands-on work begins.

Set Up Your Environment

Before you can move data between systems, every tool in the pipeline needs to be installed and connected.

In this step, you will create a project workspace, install the Python packages that power your pipeline, wire up your Snowflake warehouse and dbt project, and seed your DynamoDB table with sample e-commerce data.

In this step, get ready to:
  • Install dbt-snowflake, boto3, and PySpark into a virtual environment.
  • Create a Snowflake database and scaffold a dbt project connected to it.
  • Seed a DynamoDB table with nested e-commerce order data.
Create your virtual environment and install dependencies

A virtual environment keeps this project's packages isolated from your system Python.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir dynamodb-etl-pipeline
cd dynamodb-etl-pipeline
  • Create and activate a Python virtual environment by running:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

Make sure you ran the activate command (not just the python3 -m venv venv line). If you see a permissions error on Windows, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first.

Help me fix my virtual environment activation

  • Create your requirements.txt file by running:
cat > requirements.txt << 'EOF'
dbt-snowflake==1.11.6
boto3
pyspark
pandas
EOF

What does this file do?

  • dbt-snowflake==1.11.6 installs dbt-core and the Snowflake adapter together in one package.
  • boto3 is the AWS SDK for Python, used to read from DynamoDB.
  • pyspark gives you distributed DataFrame transformations.
  • pandas bridges PySpark and Snowflake's Python connector for loading data.
  • Install all dependencies by running:
pip install -r requirements.txt

This will take a minute or two. You will see packages downloading and installing in the terminal output.

Seeing install errors?

If you see a build error for pyspark, confirm that Java/JDK is installed by running java -version. PySpark requires Java to run.

Help me troubleshoot pip install errors

  • Before you run the version check, what output do you expect to see if dbt installed correctly?
  • Verify dbt is installed by running:
dbt --version

✔️ I see dbt-core and snowflake versions

You should see output listing Core with the installed dbt-core version and Plugins with the snowflake adapter version. You are good to continue.

ⓧ Command not found

Make sure your virtual environment is active (you see (venv) in your prompt). If not, re-run the activate command from the platform tab above.

If it is active and you still see the error, re-run pip install -r requirements.txt and check for errors in the output.

  • Verify boto3 is installed by running:
python -c "import boto3; print(boto3.__version__)"

You should see a version number printed. This confirms both Python packages are ready.

Configure Snowflake and dbt

Your pipeline needs a destination warehouse. You will create a dedicated database in Snowflake and then connect dbt to it.

  • Log in to your Snowflake account in your browser.
  • Open a new SQL worksheet by clicking + Create in the top left, then SQL Worksheet.
  • Create the project database and schema by running these SQL statements:
CREATE DATABASE ETL_PROJECT;
CREATE SCHEMA ETL_PROJECT.RAW;

What does this create?

ETL_PROJECT is a dedicated database for this pipeline. The RAW schema inside it is where your transformed data will land before dbt models it further.

You should see two success messages confirming the database and schema were created.

  • Switch back to your terminal. Make sure you are inside the dynamodb-etl-pipeline folder with your virtual environment active.
  • Scaffold your dbt project by running:
dbt init dbt_orders

What will dbt init ask me?

dbt init is interactive. It will ask which database adapter to use. Enter the number next to snowflake from the list.

It will then ask connection details (account, user, password, etc.). You can press Enter through these for now. You will configure the real connection in profiles.yml manually in the next chunk.

  • Confirm the project folder was created by running:
ls dbt_orders

You should see folders including models/, tests/, macros/, and a dbt_project.yml file.

  • Replace the contents of dbt_orders/dbt_project.yml with the following configuration:
name: "dbt_orders"
version: "1.0.0"

profile: "dbt_orders"

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

models:
  dbt_orders:
    staging:
      +materialized: view
    marts:
      +materialized: table

What does the models section do?

The models block at the bottom tells dbt how to materialize each layer. Staging models become views (lightweight, always up to date). Mart models become tables (pre-computed, fast to query).

  • Save dbt_orders/dbt_project.yml.

Now you need to tell dbt how to connect to your Snowflake account. The connection details live in a profiles.yml file.

  • Open (or create) the profiles file at ~/.dbt/profiles.yml and replace its contents with:
dbt_orders:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: [[SNOWFLAKE_ACCOUNT="your-account-identifier"]]
      user: [[SNOWFLAKE_USER="your-username"]]
      password: your-password-here
      database: ETL_PROJECT
      warehouse: [[SNOWFLAKE_WAREHOUSE="your-warehouse-name"]]
      schema: RAW
      threads: 1

Where do I find my Snowflake account identifier?

Your account identifier is the part before .snowflakecomputing.com in your Snowflake URL. For example, if your URL is abc12345.us-east-1.snowflakecomputing.com, your account is abc12345.us-east-1.

Replace your-password-here with your actual Snowflake password. The warehouse is the compute resource (often named COMPUTE_WH by default).

  • Save ~/.dbt/profiles.yml.
  • What do you think will happen when you run dbt debug? Will it connect successfully, or will it fail?
  • Test the connection by running:
cd dbt_orders
dbt debug

✔️ I see All checks passed

Your dbt project is connected to Snowflake. You should see Connection test: [OK connection ok] and All checks passed! at the bottom.

ⓧ I see a connection error

Double-check your ~/.dbt/profiles.yml values. The most common issues are:

  • Wrong account identifier (must include the region, e.g. abc12345.us-east-1).
  • Typo in the warehouse name (check it matches exactly in Snowflake).
  • Incorrect password.

Fix the values in profiles.yml, save, and run dbt debug again.

Help me troubleshoot my dbt debug connection error

Seed DynamoDB with sample data

Your pipeline needs source data to extract. You will create a DynamoDB table filled with realistic e-commerce orders that contain nested customer objects and arrays of items.

  • Navigate back to the project root by running:
cd ~/Desktop/dynamodb-etl-pipeline
  • Create a file called seed_dynamodb.py and add the table creation logic:
import boto3
import time
from decimal import Decimal

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")

table_name = "ecommerce_orders"

# Create table
try:
    table = dynamodb.create_table(
        TableName=table_name,
        KeySchema=[{"AttributeName": "order_id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "order_id", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    table.wait_until_exists()
    print(f"Table {table_name} created.")
except dynamodb.meta.client.exceptions.ResourceInUseException:
    table = dynamodb.Table(table_name)
    print(f"Table {table_name} already exists.")

What does this code do?

  • boto3.resource("dynamodb") creates a high-level connection to DynamoDB in the us-east-1 region.
  • create_table defines a table with order_id as the partition key and on-demand billing (no provisioned capacity to manage).
  • The try/except handles the case where the table already exists, so you can re-run the script safely.
  • Now add the sample order data and batch write logic below the existing code in seed_dynamodb.py:
orders = [
    {"order_id": "ORD-001", "order_date": "2024-01-15", "status": "completed", "payment_method": "credit_card", "customer": {"customer_id": "CUST-101", "name": "Alice Johnson", "email": "alice@example.com", "address": {"street": "123 Main St", "city": "Seattle", "state": "WA", "zip": "98101"}}, "items": [{"product_id": "PROD-A1", "name": "Wireless Mouse", "quantity": 2, "price": Decimal("29.99")}, {"product_id": "PROD-B2", "name": "USB-C Cable", "quantity": 3, "price": Decimal("9.99")}]},
    {"order_id": "ORD-002", "order_date": "2024-01-16", "status": "completed", "payment_method": "debit_card", "customer": {"customer_id": "CUST-102", "name": "Bob Smith", "email": "bob@example.com", "address": {"street": "456 Oak Ave", "city": "Portland", "state": "OR", "zip": "97201"}}, "items": [{"product_id": "PROD-C3", "name": "Mechanical Keyboard", "quantity": 1, "price": Decimal("89.99")}]},
    {"order_id": "ORD-003", "order_date": "2024-01-17", "status": "pending", "payment_method": "credit_card", "customer": {"customer_id": "CUST-103", "name": "Carol Davis", "email": "carol@example.com", "address": {"street": "789 Pine Rd", "city": "San Francisco", "state": "CA", "zip": "94102"}}, "items": [{"product_id": "PROD-A1", "name": "Wireless Mouse", "quantity": 1, "price": Decimal("29.99")}, {"product_id": "PROD-D4", "name": "Monitor Stand", "quantity": 1, "price": Decimal("49.99")}, {"product_id": "PROD-E5", "name": "Webcam HD", "quantity": 1, "price": Decimal("59.99")}]},
    {"order_id": "ORD-004", "order_date": "2024-01-18", "status": "shipped", "payment_method": "paypal", "customer": {"customer_id": "CUST-101", "name": "Alice Johnson", "email": "alice@example.com", "address": {"street": "123 Main St", "city": "Seattle", "state": "WA", "zip": "98101"}}, "items": [{"product_id": "PROD-F6", "name": "Laptop Sleeve", "quantity": 1, "price": Decimal("34.99")}]},
    {"order_id": "ORD-005", "order_date": "2024-01-19", "status": "completed", "payment_method": "credit_card", "customer": {"customer_id": "CUST-104", "name": "David Lee", "email": "david@example.com", "address": {"street": "321 Elm Blvd", "city": "Austin", "state": "TX", "zip": "73301"}}, "items": [{"product_id": "PROD-C3", "name": "Mechanical Keyboard", "quantity": 2, "price": Decimal("89.99")}, {"product_id": "PROD-B2", "name": "USB-C Cable", "quantity": 5, "price": Decimal("9.99")}]},
    {"order_id": "ORD-006", "order_date": "2024-01-20", "status": "cancelled", "payment_method": "credit_card", "customer": {"customer_id": "CUST-105", "name": "Eva Martinez", "email": "eva@example.com", "address": {"street": "654 Maple Dr", "city": "Denver", "state": "CO", "zip": "80201"}}, "items": [{"product_id": "PROD-E5", "name": "Webcam HD", "quantity": 1, "price": Decimal("59.99")}]},
    {"order_id": "ORD-007", "order_date": "2024-01-21", "status": "completed", "payment_method": "debit_card", "customer": {"customer_id": "CUST-102", "name": "Bob Smith", "email": "bob@example.com", "address": {"street": "456 Oak Ave", "city": "Portland", "state": "OR", "zip": "97201"}}, "items": [{"product_id": "PROD-A1", "name": "Wireless Mouse", "quantity": 1, "price": Decimal("29.99")}, {"product_id": "PROD-D4", "name": "Monitor Stand", "quantity": 2, "price": Decimal("49.99")}]},
    {"order_id": "ORD-008", "order_date": "2024-01-22", "status": "pending", "payment_method": "credit_card", "customer": {"customer_id": "CUST-106", "name": "Frank Wilson", "email": "frank@example.com", "address": {"street": "987 Cedar Ln", "city": "Chicago", "state": "IL", "zip": "60601"}}, "items": [{"product_id": "PROD-F6", "name": "Laptop Sleeve", "quantity": 3, "price": Decimal("34.99")}, {"product_id": "PROD-C3", "name": "Mechanical Keyboard", "quantity": 1, "price": Decimal("89.99")}]},
    {"order_id": "ORD-009", "order_date": "2024-01-23", "status": "completed", "payment_method": "paypal", "customer": {"customer_id": "CUST-103", "name": "Carol Davis", "email": "carol@example.com", "address": {"street": "789 Pine Rd", "city": "San Francisco", "state": "CA", "zip": "94102"}}, "items": [{"product_id": "PROD-B2", "name": "USB-C Cable", "quantity": 10, "price": Decimal("9.99")}]},
    {"order_id": "ORD-010", "order_date": "2024-01-24", "status": "shipped", "payment_method": "credit_card", "customer": {"customer_id": "CUST-107", "name": "Grace Kim", "email": "grace@example.com", "address": {"street": "147 Birch Way", "city": "Boston", "state": "MA", "zip": "02101"}}, "items": [{"product_id": "PROD-E5", "name": "Webcam HD", "quantity": 1, "price": Decimal("59.99")}, {"product_id": "PROD-A1", "name": "Wireless Mouse", "quantity": 1, "price": Decimal("29.99")}, {"product_id": "PROD-D4", "name": "Monitor Stand", "quantity": 1, "price": Decimal("49.99")}]},
]

with table.batch_writer() as batch:
    for order in orders:
        batch.put_item(Item=order)

print(f"Seeded {len(orders)} orders into {table_name}.")

What does this code do?

  • Each order has a nested customer object (with its own nested address) and an items array of products. This mimics real-world NoSQL data that does not fit neatly into rows and columns.
  • batch_writer() efficiently writes all 10 orders to DynamoDB in batches instead of one at a time.
  • Decimal is required because DynamoDB stores numbers as Decimals, not floats.
  • Save seed_dynamodb.py.
  • Run the seed script:
python seed_dynamodb.py

You should see output like:

Table ecommerce_orders created.
Seeded 10 orders into ecommerce_orders.

Seeing a credentials error?

boto3 needs AWS credentials configured. Run aws configure and enter your AWS Access Key ID, Secret Access Key, and set the default region to us-east-1. Then re-run the script.

Help me configure boto3 credentials for DynamoDB

  • Verify the data exists by running a quick scan:
python -c "import boto3; t = boto3.resource('dynamodb', region_name='us-east-1').Table('ecommerce_orders'); print(f'Item count: {t.item_count}')"

You should see Item count: 10 (or the count may show 0 initially since DynamoDB updates this metric asynchronously). If so, wait a minute and try again, or check the AWS Console.

✔️ Awesome, I've got everything!

Great. Double check you have saved both seed_dynamodb.py and dbt_orders/dbt_project.yml.

ⓧ I'd like to double check the full code

Scroll up and confirm each chunk is present in your files, in order:

  • seed_dynamodb.py chunk 1: imports, DynamoDB resource, and the try/except table creation block.
  • seed_dynamodb.py chunk 2: the orders list (10 items), batch_writer() loop, and the print statement.
  • dbt_orders/dbt_project.yml: full YAML config including the models section with staging (view) and marts (table) materialization.

Your environment is fully wired up. You have dbt talking to Snowflake, boto3 talking to DynamoDB, and 10 orders waiting to be extracted. Next up, you will pull that raw data out and see exactly what nested NoSQL documents look like in PySpark.

Extract Raw Data from DynamoDB

Your DynamoDB table is full of e-commerce orders, but the data is locked inside a NoSQL structure with nested maps and arrays. Before you can transform or analyze this data, you need to pull it out programmatically.

In this step, you will write a Python extraction function using boto3 and then load the raw results into a PySpark DataFrame to see exactly what semi-structured production data looks like.

In this step, get ready to:
  • Extract all items from DynamoDB using boto3 with pagination.
  • Load the raw data into a PySpark DataFrame and observe its nested structure.
Create the ETL pipeline file with imports and configuration

The pipeline script needs connections to three systems: AWS for extraction, PySpark for transformation, and Snowflake for loading. You will set up all the imports and connection details now so each stage can use them.

  • Create a new file called etl_pipeline.py in your project directory by adding the following imports and configuration:
import boto3
import json
from decimal import Decimal
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode, lit, sum as spark_sum
from pyspark.sql.types import (
    StructType, StructField, StringType, ArrayType, FloatType, IntegerType
)
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas

# --- Configuration ---
AWS_REGION = "us-east-1"
DYNAMODB_TABLE = "ecommerce_orders"

SNOWFLAKE_CONFIG = {
    "account": "your-account-identifier",
    "user": "your-username",
    "password": "your-password",
    "database": "ETL_PROJECT",
    "schema": "RAW",
    "warehouse": "your-warehouse-name",
}

What does this code do?

The imports pull in three tool families. boto3 talks to AWS services. pyspark provides distributed DataFrame operations. snowflake.connector handles the warehouse load.

The SNOWFLAKE_CONFIG dictionary holds your Snowflake connection details. Replace the placeholder values with the same credentials you used in your profiles.yml during setup.

  • Replace your-account-identifier, your-username, your-password, and your-warehouse-name with your actual Snowflake credentials.
  • Save etl_pipeline.py.
Write the extraction function with pagination

DynamoDB's scan() operation returns at most 1 MB of data per call. For larger tables, you must loop using LastEvaluatedKey to fetch the next page of results.

  • Add the extract_from_dynamodb() function below your configuration block:
# --- Step 1: Extract from DynamoDB ---
def extract_from_dynamodb():
    dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
    table = dynamodb.Table(DYNAMODB_TABLE)

    items = []
    response = table.scan()
    items.extend(response.get("Items", []))

    while "LastEvaluatedKey" in response:
        response = table.scan(ExclusiveStartKey=response["LastEvaluatedKey"])
        items.extend(response.get("Items", []))

    print(f"Extracted {len(items)} items from DynamoDB.")
    return items

What does this code do?

The function connects to the ecommerce_orders table and calls table.scan() to read all items. The while loop handles pagination by checking for LastEvaluatedKey in the response.

If the key exists, there are more items to fetch. The loop continues scanning from where it left off until all items are collected.

  • Add a temporary test block at the bottom of the file to run the extraction:
if __name__ == "__main__":
    print("=== EXTRACT ===")
    raw_items = extract_from_dynamodb()
  • Save etl_pipeline.py.
  • Run the script by executing this command:
python etl_pipeline.py

You should see Extracted 10 items from DynamoDB. printed to the terminal.

Not seeing 10 items?

  • Make sure your AWS credentials are configured. Run aws sts get-caller-identity to check.
  • Confirm the DynamoDB table exists by running aws dynamodb list-tables --region us-east-1 and looking for ecommerce_orders.
  • If you see 0 items, re-run python seed_dynamodb.py to populate the table.

help me debug my DynamoDB extraction

Load the raw items into a PySpark DataFrame

The extracted items are Python dictionaries with nested structures. To see exactly how PySpark interprets this nesting, you will load the raw data into a DataFrame and print its schema.

  • Replace the if __name__ block at the bottom of etl_pipeline.py with this expanded test code:
if __name__ == "__main__":
    print("=== EXTRACT ===")
    raw_items = extract_from_dynamodb()

    # Temporary: observe the raw structure in PySpark
    spark = SparkSession.builder.appName("DynamoDB_ETL").master("local[*]").getOrCreate()

    # Convert Decimal to float for Spark compatibility
    def convert_decimals(obj):
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, dict):
            return {k: convert_decimals(v) for k, v in obj.items()}
        if isinstance(obj, list):
            return [convert_decimals(i) for i in obj]
        return obj

    cleaned_items = [convert_decimals(item) for item in raw_items]

Why convert Decimals?

DynamoDB returns numeric values as Python Decimal objects. PySpark does not understand Decimal natively, so you convert them to float before creating a DataFrame.

  • Add the schema definition and DataFrame creation directly below the code you just added, still inside the if __name__ block:
    schema = StructType([
        StructField("order_id", StringType()),
        StructField("order_date", StringType()),
        StructField("status", StringType()),
        StructField("payment_method", StringType()),
        StructField("customer", StructType([
            StructField("customer_id", StringType()),
            StructField("name", StringType()),
            StructField("email", StringType()),
            StructField("address", StructType([
                StructField("street", StringType()),
                StructField("city", StringType()),
                StructField("state", StringType()),
                StructField("zip", StringType()),
            ])),
        ])),
        StructField("items", ArrayType(StructType([
            StructField("product_id", StringType()),
            StructField("name", StringType()),
            StructField("quantity", IntegerType()),
            StructField("price", FloatType()),
        ]))),
    ])

    df = spark.createDataFrame(cleaned_items, schema=schema)
    print("--- Raw Schema ---")
    df.printSchema()
    print("--- Raw Data (first 5) ---")
    df.show(5, truncate=False)
    spark.stop()

What does this schema define?

The schema tells PySpark the exact shape of each order. The customer field is a StructType (a nested map) containing another nested address struct inside it.

The items field is an ArrayType containing a struct for each product in the order. One order can have multiple items, so this is a one-to-many relationship packed into a single row.

  • Save etl_pipeline.py.
  • Before you run this, what do you think the schema printout will look like for the customer and items columns?
  • Run the script:
python etl_pipeline.py

You should see a tree-structured schema showing customer as a nested struct (with address nested further inside it) and items as an array of structs.

Seeing a Java or Spark error?

  • PySpark requires Java to be installed. Confirm Java is available by running java -version in your terminal.
  • If you see a JAVA_HOME error, set the environment variable to point to your JDK installation directory.

help me fix PySpark startup errors

Observe the nested data problem

Look at the schema and data output in your terminal. The nested structure reveals three problems that prevent direct loading into a flat Snowflake table.

  • The customer column is a nested map containing customer_id, name, email, and another nested address map. SQL cannot query into these nested levels directly.
  • The items column is an array. Each order contains a variable number of products, so one row holds multiple line items. You cannot GROUP BY product or calculate per-item totals without exploding this array into separate rows.
  • The raw DynamoDB data uses Python Decimal types for prices, which are incompatible with Snowflake's numeric columns.

Why can't you just load this into Snowflake as-is?

Snowflake tables are flat. Every row has the same columns, and every column holds a single scalar value. A nested struct or an array does not fit this model.

You need a transformation step that flattens the nested maps into top-level columns and explodes the arrays into individual rows. That is exactly what PySpark will do in the next step.

  • Now that you have confirmed the extraction works and understand the nesting problem, remove the temporary if __name__ block from etl_pipeline.py. Your file should end after the extract_from_dynamodb() function.
  • Save etl_pipeline.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py with only the imports, configuration, and extract_from_dynamodb() function.

ⓧ I'd like to double check the full code

Your etl_pipeline.py should contain exactly this:

import boto3
import json
from decimal import Decimal
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode, lit, sum as spark_sum
from pyspark.sql.types import (
    StructType, StructField, StringType, ArrayType, FloatType, IntegerType
)
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas

# --- Configuration ---
AWS_REGION = "us-east-1"
DYNAMODB_TABLE = "ecommerce_orders"

SNOWFLAKE_CONFIG = {
    "account": "your-account-identifier",
    "user": "your-username",
    "password": "your-password",
    "database": "ETL_PROJECT",
    "schema": "RAW",
    "warehouse": "your-warehouse-name",
}

Followed by the extraction function:

# --- Step 1: Extract from DynamoDB ---
def extract_from_dynamodb():
    dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
    table = dynamodb.Table(DYNAMODB_TABLE)

    items = []
    response = table.scan()
    items.extend(response.get("Items", []))

    while "LastEvaluatedKey" in response:
        response = table.scan(ExclusiveStartKey=response["LastEvaluatedKey"])
        items.extend(response.get("Items", []))

    print(f"Extracted {len(items)} items from DynamoDB.")
    return items

You have confirmed that boto3 can extract all 10 orders from DynamoDB, and you have seen firsthand how the nested maps and arrays make this data incompatible with a flat SQL table. Next up, you will use PySpark to flatten and clean this data into a shape that Snowflake can accept.

Transform with PySpark

You have raw data sitting in a PySpark DataFrame, but it is full of nested maps and arrays.

Snowflake needs flat, tabular data to run SQL against. In this step, you will flatten the nested customer objects and explode the items arrays into one row per order-item, producing a clean DataFrame ready for loading.

In this step, get ready to:
  • Flatten nested customer and address fields into top-level columns.
  • Explode the items array so each product becomes its own row.
  • Derive a calculated line_total column and convert the result to Pandas.
Set up PySpark and handle Decimal conversion

DynamoDB stores numbers as Python Decimal objects, which PySpark does not accept natively. You need a helper function that recursively converts every Decimal to a float before PySpark can process the data.

  • In etl_pipeline.py, find the blank line after the extract_from_dynamodb() function. Add the following code below it:
# --- Step 2: Transform with PySpark ---
def transform_with_pyspark(raw_items):
    spark = SparkSession.builder.appName("DynamoDB_ETL").master("local[*]").getOrCreate()

    # Convert Decimal to float for Spark compatibility
    def convert_decimals(obj):
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, dict):
            return {k: convert_decimals(v) for k, v in obj.items()}
        if isinstance(obj, list):
            return [convert_decimals(i) for i in obj]
        return obj

    cleaned_items = [convert_decimals(item) for item in raw_items]

What does this code do?

SparkSession.builder creates a local Spark instance on your machine. The convert_decimals helper walks through every item recursively, turning Decimal values into floats that Spark can handle.

cleaned_items is the list of DynamoDB records with all Decimals converted, ready to become a Spark DataFrame.

  • Directly below the cleaned_items line, add the explicit schema and DataFrame creation:
    schema = StructType([
        StructField("order_id", StringType()),
        StructField("order_date", StringType()),
        StructField("status", StringType()),
        StructField("payment_method", StringType()),
        StructField("customer", StructType([
            StructField("customer_id", StringType()),
            StructField("name", StringType()),
            StructField("email", StringType()),
            StructField("address", StructType([
                StructField("street", StringType()),
                StructField("city", StringType()),
                StructField("state", StringType()),
                StructField("zip", StringType()),
            ])),
        ])),
        StructField("items", ArrayType(StructType([
            StructField("product_id", StringType()),
            StructField("name", StringType()),
            StructField("quantity", IntegerType()),
            StructField("price", FloatType()),
        ]))),
    ])

    df = spark.createDataFrame(cleaned_items, schema=schema)

    print("--- Raw Schema ---")
    df.printSchema()
    print("--- Raw Data (first 5) ---")
    df.show(5, truncate=False)

Why define an explicit schema?

PySpark can infer schemas from data, but DynamoDB items have inconsistent shapes (some fields may be missing). An explicit StructType schema tells Spark exactly what to expect: a nested customer struct with an address sub-struct, and an items array of structs.

The printSchema() and show() calls let you see the raw nested structure before flattening.

  • At the very bottom of etl_pipeline.py, add this main block so you can run the pipeline:
# --- Main Pipeline ---
if __name__ == "__main__":
    print("=== EXTRACT ===")
    raw_items = extract_from_dynamodb()

    print("\n=== TRANSFORM ===")
    pandas_df = transform_with_pyspark(raw_items)
  • Save etl_pipeline.py and run it:
python etl_pipeline.py

You should see the raw nested schema printed, showing customer as a struct and items as an array. The raw data preview shows all the nesting that needs flattening.

Seeing a Java or Spark error?

PySpark requires Java/JDK to run. If you see a Java-related error, confirm Java is on your PATH by running java -version in your terminal. If Spark cannot find Java, set your JAVA_HOME environment variable to point to your JDK installation directory.

help me fix a PySpark Java error when running my ETL script

Flatten customer fields and explode the items array

The raw DataFrame has a nested customer struct and an items array. You need to pull customer fields up to top-level columns and use explode() to turn each item in the array into its own row.

  • In etl_pipeline.py, find the df.show(5, truncate=False) line inside transform_with_pyspark. Add the following code directly below it:
    # Flatten customer fields and explode items array
    df_flat = df.select(
        col("order_id"),
        col("order_date"),
        col("status"),
        col("payment_method"),
        col("customer.customer_id").alias("customer_id"),
        col("customer.name").alias("customer_name"),
        col("customer.email").alias("customer_email"),
        col("customer.address.city").alias("customer_city"),
        col("customer.address.state").alias("customer_state"),
        explode(col("items")).alias("item"),
    )

What does this code do?

Dot notation like col("customer.customer_id") reaches inside the nested struct to pull fields up to the top level. Each .alias() call gives the extracted field a clean column name.

explode(col("items")) creates one row for every element in the items array. An order with 3 items becomes 3 rows, each carrying the same order-level fields.

Extract item fields and derive line_total

After the explode, each row has an item struct column containing product details. You need to extract those fields and calculate a line_total for each order-item row.

  • Directly below the df_flat block you just added, add the item extraction:
    # Extract item fields into top-level columns
    df_transformed = df_flat.select(
        col("order_id"),
        col("order_date"),
        col("status"),
        col("payment_method"),
        col("customer_id"),
        col("customer_name"),
        col("customer_email"),
        col("customer_city"),
        col("customer_state"),
        col("item.product_id").alias("product_id"),
        col("item.name").alias("product_name"),
        col("item.quantity").alias("quantity"),
        col("item.price").alias("unit_price"),
    )
  • Directly below the df_transformed select, add the derived column, output, and return:
    # Derive line_total
    df_transformed = df_transformed.withColumn(
        "line_total", col("quantity") * col("unit_price")
    )

    print("--- Transformed Schema ---")
    df_transformed.printSchema()
    print("--- Transformed Data ---")
    df_transformed.show(truncate=False)

    # Convert to Pandas for Snowflake loading
    pandas_df = df_transformed.toPandas()

    spark.stop()
    return pandas_df

What does this code do?

The second select() pulls each item field (product_id, product_name, quantity, unit_price) out of the item struct into proper columns.

withColumn adds a calculated line_total column (quantity multiplied by unit_price). Finally, toPandas() converts the Spark DataFrame into a Pandas DataFrame that the Snowflake connector can load directly.

Before you run this, what do you think the output will look like? Will the data still be nested, or flat?

  • Save etl_pipeline.py and run it again:
python etl_pipeline.py

You should see a clean, flat DataFrame printed with columns like order_id, customer_id, customer_name, product_name, quantity, unit_price, and line_total. No more nested structs or arrays.

Seeing an AnalysisException or column not found?

Double check that the column names in your col() calls match exactly: customer.customer_id (not customer_id by itself for the nested field). Also confirm item.product_id uses the alias item from the explode, not items.

help me debug a PySpark AnalysisException in my transform function

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py and that your terminal shows the flat transformed output.

ⓧ I'd like to double check the full code

Your etl_pipeline.py should contain these sections in order. Scroll up through your file and confirm each is present:

  • Import block (boto3, json, Decimal, SparkSession, col, explode, lit, spark_sum, StructType, StructField, StringType, ArrayType, FloatType, IntegerType, snowflake.connector, write_pandas).
  • Configuration block with AWS_REGION, DYNAMODB_TABLE, and SNOWFLAKE_CONFIG.
  • extract_from_dynamodb() function using table.scan() with pagination.
  • transform_with_pyspark(raw_items) function containing: SparkSession creation, convert_decimals helper, explicit schema, createDataFrame, raw printSchema/show, df_flat with explode, df_transformed select, withColumn for line_total, transformed printSchema/show, toPandas(), spark.stop(), return pandas_df.
  • Main block at the bottom calling extract_from_dynamodb() and transform_with_pyspark(raw_items).

Your transform function turns messy nested documents into a clean, flat table. Next up, you will load this data into Snowflake and start querying it with SQL.

Load to Snowflake and Query with SQL

You have clean, flat data sitting in a Pandas DataFrame. Now it needs to land in Snowflake so you can query it with SQL and so dbt can model it downstream.

In this step, you will write the load function, run the full pipeline end-to-end, and then query the raw data directly. You will also discover a limitation that motivates the next step.

In this step, get ready to:
  • Write a load function that pushes your DataFrame into Snowflake.
  • Run the full ETL pipeline from extract through load.
  • Query the raw table with SQL and observe what is missing.
Write the load function and run the pipeline

Snowflake expects uppercase column names. The write_pandas helper from the Snowflake connector handles staging and bulk loading in a single call.

  • In etl_pipeline.py, scroll to the bottom of your transform_with_pyspark function.
  • Add the following load_to_snowflake function below it:
# --- Step 3: Load to Snowflake ---
def load_to_snowflake(pandas_df):
    conn = snowflake.connector.connect(**SNOWFLAKE_CONFIG)

    # Uppercase column names for Snowflake
    pandas_df.columns = [c.upper() for c in pandas_df.columns]

    success, nchunks, nrows, _ = write_pandas(
        conn, pandas_df, "ORDERS_FLAT", auto_create_table=True
    )

    print(f"Loaded {nrows} rows into ORDERS_FLAT ({nchunks} chunks). Success: {success}")
    conn.close()

What does this code do?

This function connects to Snowflake using your saved config, then uppercases every column name so Snowflake can match them correctly.

  • auto_create_table=True tells Snowflake to create the ORDERS_FLAT table automatically based on the DataFrame columns.
  • write_pandas uses Parquet files and a temporary stage under the hood for efficient bulk loading.
  • Now add the main pipeline block at the very bottom of etl_pipeline.py, below the load_to_snowflake function:
# --- Main Pipeline ---
if __name__ == "__main__":
    print("=== EXTRACT ===")
    raw_items = extract_from_dynamodb()

    print("\n=== TRANSFORM ===")
    pandas_df = transform_with_pyspark(raw_items)

    print("\n=== LOAD ===")
    load_to_snowflake(pandas_df)

    print("\n=== PIPELINE COMPLETE ===")

What does this code do?

The if __name__ == "__main__" block runs when you execute the script directly. It calls your three functions in sequence: extract, transform, load.

The print statements act as progress markers so you can see each stage start in the terminal.

  • Save etl_pipeline.py.

Before you run the pipeline, what do you expect to see as the final output? Think about how many rows will be created from 10 orders with multiple items each.

  • Run the full pipeline by executing this command:
python etl_pipeline.py

You should see output showing all three stages complete. The final line reports 18 rows loaded into ORDERS_FLAT. That is because each order item became its own row after the explode transformation.

Seeing a connection error?

  • Check that SNOWFLAKE_CONFIG in etl_pipeline.py has your real account identifier, username, password, and warehouse name filled in.
  • Make sure your Snowflake warehouse is not suspended. You can resume it in Snowsight under **Warehouses**.
  • Confirm your virtual environment is activated and all packages from requirements.txt are installed.

Help me debug my Snowflake connection error in etl_pipeline.py.

Query the raw data and notice the gap

Your data is in Snowflake. Now you can query it with SQL to verify the load and explore what you have.

  • Open Snowflake Snowsight in your browser and navigate to **Worksheets**.
  • Create a new worksheet by clicking the **+** button.
  • Run this query to verify the row count:
SELECT COUNT(*) AS total_rows FROM ETL_PROJECT.RAW.ORDERS_FLAT;

You should see 18 as the result. That confirms all 18 order-item rows landed successfully.

  • Run this query to preview a few rows:
SELECT * FROM ETL_PROJECT.RAW.ORDERS_FLAT LIMIT 10;

You should see flat rows with columns like ORDER_ID, CUSTOMER_NAME, PRODUCT_NAME, and LINE_TOTAL. The nested DynamoDB structure is gone.

  • Now run an aggregation to see spending by customer:
SELECT
    CUSTOMER_ID,
    CUSTOMER_NAME,
    SUM(LINE_TOTAL) AS total_spend
FROM ETL_PROJECT.RAW.ORDERS_FLAT
GROUP BY CUSTOMER_ID, CUSTOMER_NAME
ORDER BY total_spend DESC;
  • Run one more query to see order counts by status:
SELECT
    STATUS,
    COUNT(DISTINCT ORDER_ID) AS order_count
FROM ETL_PROJECT.RAW.ORDERS_FLAT
GROUP BY STATUS;

These queries work. You can answer basic business questions right now. But look at what you are working with.

What is the problem here?

You have one giant flat table with no separation of concerns. Customer data, order data, and item data are all mashed together in a single table.

There are no data quality checks. If a null customer_id snuck through or a duplicate order appeared, nothing would catch it.

There is no documentation describing what each column means or what values are valid. This is the problem that dbt solves in the next step.

  • Create a file called queries.sql in your project directory to save these queries for reference:
-- SQL Queries for ETL Pipeline Verification

-- Check row count
SELECT COUNT(*) AS total_rows FROM ETL_PROJECT.RAW.ORDERS_FLAT;

-- Preview the data
SELECT * FROM ETL_PROJECT.RAW.ORDERS_FLAT LIMIT 10;

-- Quick aggregation on raw table: spend by customer
SELECT
    CUSTOMER_ID,
    CUSTOMER_NAME,
    SUM(LINE_TOTAL) AS total_spend
FROM ETL_PROJECT.RAW.ORDERS_FLAT
GROUP BY CUSTOMER_ID, CUSTOMER_NAME
ORDER BY total_spend DESC;

-- Order counts by status
SELECT
    STATUS,
    COUNT(DISTINCT ORDER_ID) AS order_count
FROM ETL_PROJECT.RAW.ORDERS_FLAT
GROUP BY STATUS;

Why save queries in a file?

Keeping your SQL queries in version control makes them repeatable and shareable. Anyone on your team can re-run the same verification after a pipeline run.

  • Save queries.sql.
  • Confirm the file exists by running:
ls queries.sql

You should see queries.sql listed in the output.

✔️ Awesome, I've got everything!

Great. Double check you have saved both etl_pipeline.py and queries.sql.

ⓧ I'd like to double check the full code

Your etl_pipeline.py should now contain these sections in order. Scroll up and confirm each one is present:

  • Imports (boto3, json, Decimal, PySpark, snowflake.connector, write_pandas).
  • Configuration block with AWS_REGION, DYNAMODB_TABLE, and SNOWFLAKE_CONFIG.
  • extract_from_dynamodb() function with scan pagination.
  • transform_with_pyspark() function with schema, flatten, explode, and line_total.
  • load_to_snowflake() function with uppercase columns and write_pandas.
  • Main pipeline block calling all three functions in sequence.

Your queries.sql should contain all four queries (COUNT, preview, spend by customer, order counts by status).

Your data is in Snowflake and you can query it. But one flat table with no tests or documentation is fragile. Next up, you will use dbt to build a proper analytics layer with staging models, mart models, and data quality checks.

Model with dbt and Run Analytics Queries

You loaded your transformed data into Snowflake and proved you can query it with SQL.

However, you discovered a gap: the raw table has no separation of concerns, no data quality checks, and no documentation. dbt solves this by letting you build tested, documented transformation layers using pure SQL.

In this step, get ready to:
  • Create staging and mart dbt models with SQL.
  • Add source definitions and data quality tests.
  • Run analytical SQL queries against the finished mart models.
Create staging and mart models

In dbt, a model is a SQL SELECT statement saved as a file. When you run dbt run, dbt executes each model and materializes the result as a view or table in your warehouse.

  • Create the staging and marts directories inside your dbt project by running these commands:
cd dbt_orders
mkdir -p models/staging models/marts
  • Create a new file called models/sources.yml by pasting the following code. This tells dbt where your raw data lives:
version: 2

sources:
  - name: raw
    database: ETL_PROJECT
    schema: RAW
    tables:
      - name: orders_flat
        description: "Flattened e-commerce order data loaded from DynamoDB via PySpark ETL pipeline"

What does this file do?

The sources.yml file declares external tables that dbt does not manage but that your models read from. The {{ source('raw', 'orders_flat') }} function in your models will resolve to ETL_PROJECT.RAW.ORDERS_FLAT based on this definition.

  • Save models/sources.yml.
  • Create the staging model models/staging/stg_orders.sql by pasting the following code:
with source as (
    select * from {{ source('raw', 'orders_flat') }}
),

staged as (
    select
        order_id,
        order_date::date as order_date,
        status,
        payment_method,
        customer_id,
        customer_name,
        customer_email,
        customer_city,
        customer_state,
        product_id,
        product_name,
        quantity::int as quantity,
        unit_price::float as unit_price,
        line_total::float as line_total
    from source
)

select * from staged

What does this model do?

The staging model is a thin cleaning layer. It selects from the raw source, casts order_date to a proper date type, and ensures quantity, unit_price, and line_total are the correct numeric types.

Downstream mart models will reference this staging model using {{ ref('stg_orders') }} instead of querying the raw table directly. This separation means if your raw schema changes, you only fix the staging model.

  • Save models/staging/stg_orders.sql.
  • Confirm both files exist by running:
ls models/sources.yml models/staging/stg_orders.sql

You should see both file paths printed without errors.

  • Create the fact model models/marts/fct_order_items.sql by pasting the following code:
with stg as (
    select * from {{ ref('stg_orders') }}
)

select
    order_id,
    product_id,
    order_date,
    status,
    payment_method,
    customer_id,
    product_name,
    quantity,
    unit_price,
    line_total
from stg

What does the fact table do?

The fct_order_items model is the core analytical table at the order-item grain. Each row represents one product within one order. It references the staging model via {{ ref('stg_orders') }}, so dbt knows to build staging first.

  • Save models/marts/fct_order_items.sql.
  • Create the customer dimension model models/marts/dim_customers.sql by pasting the following code:
with stg as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select
        customer_id,
        customer_name,
        customer_email,
        customer_city,
        customer_state,
        count(distinct order_id) as total_orders,
        sum(line_total) as total_spend
    from stg
    group by
        customer_id,
        customer_name,
        customer_email,
        customer_city,
        customer_state
)

select * from customers

What does the dimension table do?

The dim_customers model deduplicates customers and pre-aggregates their total_orders and total_spend. Analysts can join this dimension to the fact table or query it directly for customer-level insights.

  • Save models/marts/dim_customers.sql.
  • Confirm the mart files exist by running:
ls models/marts/

You should see dim_customers.sql and fct_order_items.sql listed.

✔️ Awesome, I've got everything!

Great. Make sure all four files are saved before continuing.

ⓧ I'd like to double check the full code

You should have these files created so far. Scroll up and confirm each is present:

  • models/sources.yml (source definition pointing to ETL_PROJECT.RAW.ORDERS_FLAT).
  • models/staging/stg_orders.sql (CTE selecting from source with type casts).
  • models/marts/fct_order_items.sql (selects from stg_orders, 10 columns).
  • models/marts/dim_customers.sql (groups by customer fields, aggregates total_orders and total_spend).

Missing files or directories?

  • Make sure you ran mkdir -p models/staging models/marts from inside the dbt_orders directory.
  • If ls models/staging/ shows nothing, double-check that you created the file in the correct path (not the project root).

Why are my dbt model files not showing up in the correct directories?

Add tests, run dbt, and query the mart models

Your models exist as SQL files on disk, but they have not been compiled or materialized yet. Before running them, you will add data quality tests so dbt checks the output after every build.

  • Create models/marts/schema.yml by pasting the following code:
version: 2

models:
  - name: fct_order_items
    description: "Fact table at the order-item grain with financial metrics"
    columns:
      - name: order_id
        tests:
          - not_null
      - name: product_id
        tests:
          - not_null
      - name: customer_id
        tests:
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ["completed", "pending", "shipped", "cancelled"]

  - name: dim_customers
    description: "Customer dimension with aggregated order metrics"
    columns:
      - name: customer_id
        tests:
          - unique
          - not_null

What do these tests check?

  • not_null on order_id, product_id, and customer_id ensures no row is missing a key identifier.
  • accepted_values on status catches any unexpected order status that might indicate bad source data.
  • unique on customer_id in dim_customers proves the deduplication logic works correctly.
  • Save models/marts/schema.yml.

Before you run this, do you think dbt will successfully materialize all three models on the first try? What could go wrong?

  • Materialize all models in Snowflake by running:
dbt run

You should see dbt build each model in dependency order: stg_orders first (as a view), then fct_order_items and dim_customers (as tables). The output ends with:

Completed successfully

Done. PASS=3 WARN=0 ERROR=0 SKIP=0 TOTAL=3

Seeing compilation errors?

  • If dbt says it cannot find the source raw, confirm models/sources.yml exists and uses name: raw (lowercase).
  • If dbt reports a Snowflake connection error, run dbt debug to verify your profiles.yml credentials are still valid.
  • If a model fails with a SQL compilation error, check for typos in column names against the ORDERS_FLAT table schema.

Why is my dbt run failing with a compilation or connection error?

Now that your models are materialized, do you expect all the data quality tests to pass? Why or why not?

  • Validate data quality by running:
dbt test

You should see each test pass individually, followed by the summary:

Completed successfully

Done. PASS=6 WARN=0 ERROR=0 SKIP=0 TOTAL=6

What just happened?

dbt generated a SQL query for each test. For not_null, it checked if any rows had a null value in that column. For unique, it looked for duplicate customer_id values. For accepted_values, it searched for status values outside the allowed list.

Zero failing rows means every test passed. Your data is clean.

Your analytics layer is live. Time to prove it delivers real business answers by querying the mart models directly.

  • Run this query in Snowflake Snowsight (or any SQL client connected to your ETL_PROJECT database) to find your top customers by spend:
SELECT
    CUSTOMER_NAME,
    TOTAL_ORDERS,
    TOTAL_SPEND
FROM ETL_PROJECT.RAW.DIM_CUSTOMERS
ORDER BY TOTAL_SPEND DESC
LIMIT 5;

You should see a ranked list of customers with their order counts and total spend. Alice Johnson and David Lee should be near the top.

  • Run this query to see revenue by product:
SELECT
    PRODUCT_NAME,
    SUM(LINE_TOTAL) AS revenue,
    SUM(QUANTITY) AS units_sold
FROM ETL_PROJECT.RAW.FCT_ORDER_ITEMS
GROUP BY PRODUCT_NAME
ORDER BY revenue DESC;

You should see each product ranked by revenue. Mechanical Keyboard and Monitor Stand should appear near the top.

  • Run this query to see orders grouped by status:
SELECT
    STATUS,
    COUNT(DISTINCT ORDER_ID) AS order_count,
    SUM(LINE_TOTAL) AS total_revenue
FROM ETL_PROJECT.RAW.FCT_ORDER_ITEMS
GROUP BY STATUS
ORDER BY total_revenue DESC;

You should see four status categories (completed, pending, shipped, cancelled) with their order counts and revenue totals.

Why query the mart models instead of the raw table?

These queries hit DIM_CUSTOMERS and FCT_ORDER_ITEMS instead of ORDERS_FLAT. The mart models give you pre-aggregated customer metrics, guaranteed-clean types, and tested data quality. If someone adds a bad row to the raw table, your dbt tests will catch it before it pollutes your analytics.

Secret mission

Incremental Models and Source Freshness

Your pipeline works, but it reprocesses every single row each time it runs. In production, that wastes compute and time. Can you make dbt smart enough to only process new data, and alert you when upstream data stops flowing?

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 DynamoDB within free tier limits and your existing Snowflake account credits, so no unexpected charges will accumulate as long as your Snowflake warehouse is suspended when idle.

Resources you used:

  • AWS DynamoDB table ecommerce_orders in us-east-1 (free tier, no ongoing cost under 25 GB).
  • Snowflake database ETL_PROJECT with tables (ORDERS_FLAT, FCT_ORDER_ITEMS, DIM_CUSTOMERS), view (STG_ORDERS), and schema (RAW).
  • Local project files (etl_pipeline.py, seed_dynamodb.py, queries.sql, requirements.txt, dbt_orders/ directory) and Python virtual environment.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with the pipeline or running additional queries.

  • Your DynamoDB table stays within the AWS free tier (under 25 GB storage and 25 RCU/WCU).
  • Your Snowflake warehouse auto-suspends after a period of inactivity, so it stops consuming credits when idle.
  • Your local project files and virtual environment remain on your Desktop.

Pause - I'll come back to this later

Suspend your Snowflake warehouse manually to ensure zero credit usage while you are away. DynamoDB requires no action since it stays within free tier.

  • Run this SQL command in the Snowflake worksheet to suspend your warehouse:
ALTER WAREHOUSE IF EXISTS your-warehouse-name SUSPEND;
  • Your DynamoDB table remains at no cost within the free tier.
  • Your local files stay on disk, ready for you to resume.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

  • Delete the DynamoDB table by running this AWS CLI command in your terminal:
aws dynamodb delete-table --table-name ecommerce_orders

No AWS CLI installed?

If you do not have the AWS CLI, open the DynamoDB console in your browser. Select the ecommerce_orders table, click Actions, then click Delete table.

  • Drop the Snowflake database by running this SQL command in your Snowflake worksheet:
DROP DATABASE ETL_PROJECT;
  • Remove your local project files. On macOS or Linux, run:
cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]
  • On Windows, run:
cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]
  • Deactivate and remove your Python virtual environment (if it was inside the project folder, the step above already removed it).

Nice Work!

Nice Work!

Nice work! You just built a production-style ETL pipeline that bridges the NoSQL and SQL worlds, from messy nested documents in DynamoDB all the way to tested, documented analytics models in Snowflake.

You've learned how to:

  • Built a complete ETL pipeline that extracts nested data from DynamoDB with boto3 pagination, transforms it into flat tables using PySpark explode and struct extraction, and loads it into Snowflake with write_pandas.
  • Created a dbt project with staging models, mart models, source definitions, and data quality tests that validate your data on every run.
  • Wrote analytical SQL queries against finished mart models to answer real business questions like top customers by spend, revenue by product, and order trends over time.
  • Secret Mission: Implemented incremental materialization and source freshness monitoring for production-readiness.

Ready to quiz yourself?