E-Commerce Order Analytics Pipeline

Introduction

30 Second Summary

Every time someone clicks "Buy Now" on an e-commerce site, that order becomes a row in a database. At scale, companies process millions of these rows daily to answer questions like "which products are trending?" and "which customers are churning?"

In this project, you will build a production-grade batch data pipeline that ingests 5M+ e-commerce order events into an S3 data lake, transforms them with PySpark, and loads a Snowflake star schema ready for analyst queries.

What You'll Build

When you run docker compose up and trigger the DAG, you watch Airflow orchestrate the full pipeline end-to-end: data lands in S3, Spark deduplicates and enriches it, curated Parquet files appear in the curated layer, and Snowflake loads the dimensional model you can query immediately.

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

• A data generator that creates 5M+ realistic order events with intentional duplicates and late-arriving records, then lands them as date-partitioned Parquet in your S3 data lake.
• A PySpark transformation job that deduplicates via window functions, enriches orders with product data, handles late arrivals by re-partitioning to the correct event date, and writes a clean curated layer you can verify has ~15% fewer rows than the raw input.
• A Snowflake star schema where you can query revenue by product category, customer lifetime value, and daily order trends, powered by SCD Type 2 customer tracking and idempotent load patterns.
• A Dockerized Airflow DAG that orchestrates the full pipeline with retries, incremental processing, and catchup backfill, deployable with a single command.
• Secret Mission: Add data quality gates that validate your curated data and halt the pipeline before bad records reach the warehouse.

Are there any prerequisites?

This project assumes intermediate Python skills, basic SQL knowledge, and an understanding of relational databases.

You also need an AWS account with the CLI configured and Docker Desktop installed. The project uses a free Snowflake trial ($400 in credits, no credit card required).

Before We Start

Before any hands-on work, let's make sure you're clear on what you're building and the problem it solves. That clarity will guide every decision across the five steps ahead.

Set Up the Project Structure

Your tools are installed and your cloud accounts are ready. Now you need a project layout that tells Docker how to build your Airflow environment, what Python packages to install, and where to find your pipeline code.

In this step, you will create the folder structure, dependency files, and Docker configuration that wire everything together. By the end, a single docker compose config command will prove your setup is valid.

In this step, get ready to:

• Create the directory layout for the pipeline.
• Configure Docker to run Airflow with PySpark.
• Set up environment variables for AWS and Snowflake credentials.

Create the project directories

A Dockerized Airflow project expects a specific folder layout. Each folder has a clear responsibility: DAG definitions, source code, SQL scripts, and runtime logs.

• Create the four project subdirectories by running this command:

mkdir src dags snowflake logs

What do these folders hold?

• src/ holds your Python pipeline scripts (data generator, Spark transform, Snowflake loader).
• dags/ holds Airflow DAG definitions that orchestrate the pipeline.
• snowflake/ holds SQL DDL scripts for the dimensional model.
• logs/ is where Airflow writes its task logs at runtime.

• Confirm the directories exist by running:

ls

You should see src, dags, snowflake, and logs listed in the output.

Create the Python dependency files

This project uses two separate requirements files. One defines what your local virtual environment needs. The other defines what the Docker container installs for Airflow and Spark.

• Create requirements.txt for local development by adding a new file with this content:

boto3>=1.35.0
pyarrow>=15.0.0
pandas>=2.0.0
faker>=28.0.0
snowflake-connector-python>=3.12.0

What does this file do?

This pins the minimum versions of each library your local scripts need. You already installed these into your virtual environment in the previous step. The file documents the dependencies so anyone cloning the project can reproduce the environment.

• Create requirements-airflow.txt for the Docker container by adding a new file with this content:

pyspark==4.1.2
boto3>=1.35.0
pyarrow>=15.0.0
snowflake-connector-python>=3.12.0
faker>=28.0.0
pandas>=2.0.0

Why a separate file for Docker?

The Docker container also needs pyspark==4.1.2 pinned to an exact version. PySpark is not installed locally because it runs inside the container where Java is available. Keeping the files separate avoids installing unnecessary packages in each environment.

• Save both files and confirm they exist by running:

ls *.txt

You should see requirements-airflow.txt and requirements.txt in the output.

Don't see both files?

Make sure you created both files in the project root directory, not inside one of the subdirectories. Run pwd to confirm you are in the correct folder.

Help me find my requirements files.

Create the environment variables file

Docker Compose reads a .env file to inject credentials into the containers at runtime. This keeps secrets out of your code and version control.

• Create a .env file in the project root with the following content, replacing each placeholder with your actual values:

AWS_ACCESS_KEY_ID=your-access-key-here
AWS_SECRET_ACCESS_KEY=your-secret-key-here
AWS_DEFAULT_REGION=us-east-1
S3_BUCKET=ecommerce-pipeline-dev
SNOWFLAKE_ACCOUNT=your-account-identifier-here
SNOWFLAKE_USER=your-username-here
SNOWFLAKE_PASSWORD=your-password-here

Where do I find these values?

• AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are the same credentials you used with aws configure in Step 1.
• S3_BUCKET is your bucket name (e.g. ecommerce-pipeline-jd-dev). Replace ecommerce-pipeline-dev with the name you created earlier.
• SNOWFLAKE_ACCOUNT is the account identifier from your Snowflake URL (the part before .snowflakecomputing.com).
• SNOWFLAKE_USER and SNOWFLAKE_PASSWORD are the credentials you created at signup.

Never commit this file

Your .env file contains live credentials. If you later add version control, add .env to your .gitignore immediately.

• Confirm the file was created by running:

ls -a .env

You should see .env in the output. Files starting with a dot are hidden by default, which is why you need the -a flag.

Build the Docker configuration

Airflow runs inside Docker containers that need Java (for PySpark) and your Python packages. The Dockerfile defines the custom image, and docker-compose.yml defines how all the services start together.

• Create a file called Dockerfile (no extension) in the project root with this content:

FROM apache/airflow:2.11.0-python3.12

USER root

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
    openjdk-17-jre-headless \
    procps \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

ENV JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
ENV PATH="${JAVA_HOME}/bin:${PATH}"

USER airflow

COPY requirements-airflow.txt /tmp/requirements-airflow.txt
RUN pip install --no-cache-dir -r /tmp/requirements-airflow.txt

What does this Dockerfile do?

• FROM apache/airflow:2.11.0-python3.12 starts from the official Apache Airflow 2.11.0 image with Python 3.12.
• The RUN apt-get block installs Java 17, which PySpark 4.1.2 requires to run.
• COPY and pip install add your project dependencies from requirements-airflow.txt.

• Confirm the Dockerfile exists by running:

ls Dockerfile

You should see Dockerfile listed.

Now you will build docker-compose.yml section by section. This file defines four services: a Postgres database for Airflow metadata, an initializer, the web UI, and the scheduler.

• Create docker-compose.yml and add the shared configuration anchor at the top:

version: '3.8'

x-airflow-common:
  &airflow-common
  build:
    context: .
    dockerfile: Dockerfile
  environment:
    &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
    AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
    AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
    AWS_DEFAULT_REGION: us-east-1
    SNOWFLAKE_ACCOUNT: ${SNOWFLAKE_ACCOUNT}
    SNOWFLAKE_USER: ${SNOWFLAKE_USER}
    SNOWFLAKE_PASSWORD: ${SNOWFLAKE_PASSWORD}
    S3_BUCKET: ${S3_BUCKET}
  volumes:
    - ./dags:/opt/airflow/dags
    - ./src:/opt/airflow/src
    - ./logs:/opt/airflow/logs
  depends_on:
    postgres:
      condition: service_healthy

What is x-airflow-common?

This YAML anchor (&airflow-common) defines configuration that multiple services share. Instead of repeating the build context, environment variables, and volume mounts in every service, each service references this anchor with <<: *airflow-common.

The ${...} syntax pulls values from the .env file you just created. The volumes section mounts your local dags/ and src/ folders into the container so Airflow can access your code.

• Add the Postgres database service below what you just pasted:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres-db-volume:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5
      start_period: 5s
    restart: always

What does the Postgres service do?

postgres stores Airflow's internal metadata (DAG runs, task states, connections). The healthcheck ensures other services wait until Postgres is ready before starting.

• Add the Airflow initializer service below the postgres service:

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    command:
      - -c
      - |
        airflow db migrate
        airflow users create \
          --username airflow \
          --firstname Airflow \
          --lastname Admin \
          --role Admin \
          --email admin@example.com \
          --password airflow
    depends_on:
      postgres:
        condition: service_healthy

What does airflow-init do?

This service runs once at startup to create the database schema and an admin user (username: airflow, password: airflow). After the initial setup, you will use these credentials to log into the Airflow web UI.

• Add the webserver, scheduler, and volume definition below the airflow-init service:

  airflow-webserver:
    <<: *airflow-common
    command: webserver
    ports:
      - "8080:8080"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    restart: always

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler
    restart: always

volumes:
  postgres-db-volume:

What do the webserver and scheduler do?

• airflow-webserver serves the Airflow UI at localhost:8080 where you monitor and trigger DAGs.
• airflow-scheduler watches for DAG schedules and triggers task execution.
• postgres-db-volume is a named Docker volume that persists the Postgres data across container restarts.

Before you run the validation, predict what you expect to see. If the YAML is valid and the .env values resolve, the command should complete silently with no output.

• Validate the entire Docker setup by running:

docker compose config --quiet

If the command completes with no output, your configuration is valid. An error message means something is wrong with your YAML syntax or a referenced file is missing.

Seeing a YAML parsing error?

• YAML is whitespace-sensitive. Make sure you used spaces (not tabs) for indentation, and that each nested level is exactly 2 spaces deeper than its parent.
• Check that the services: key is at the root level (no leading spaces), and that each service name (postgres, airflow-init, etc.) is indented 2 spaces under services:.
• Make sure .env exists in the same directory as docker-compose.yml. Run ls -a to confirm.

Help me fix my docker-compose.yml error.

✔️ Awesome, I've got everything!

Great. Double check you have saved all files before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm you have these files in your project root, each matching the content shown above:

• requirements.txt (5 lines, local dependencies).
• requirements-airflow.txt (6 lines, includes pyspark==4.1.2).
• .env (7 lines, your AWS and Snowflake credentials).
• Dockerfile (16 lines, installs Java 17 and pip packages).
• docker-compose.yml (all four sections joined together as one file).

Run docker compose config --quiet one more time after any corrections. No output means everything is valid.

Your project skeleton is in place. Next up, you will write the data generator that fills your S3 bucket with millions of realistic e-commerce order events.

Generate and Land Raw Data in S3

Your environment is ready with Docker, AWS CLI, and Snowflake configured. Now you need data to process.

A production batch pipeline needs realistic source data that includes real-world problems: duplicate records from retried ingestion, late-arriving events that land after their business date, and nullable fields. In this step, you will build a Python generator that creates 5M+ e-commerce order events and lands them as partitioned Parquet files in your S3 data lake.

In this step, get ready to:

• Build a data generator that creates products, customers, and orders with intentional duplicates and late arrivals.
• Upload all generated data as Parquet files to S3 with date-based partitioning.
• Verify the raw layer of your data lake has the correct partition structure.

Build the data generation functions

The generator script uses the Faker library to create realistic-looking e-commerce data across 30 days of simulated history. Each day produces approximately 170,000 orders.

• Create src/generate_data.py with the following imports and seed configuration:

# Standard library for unique IDs, randomization, and date math
import uuid
import random
from datetime import datetime, timedelta

# Data processing and S3 upload libraries
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import boto3
from faker import Faker

# Fixed seeds ensure reproducible data across runs
fake = Faker()
Faker.seed(42)
random.seed(42)

What does this code do?

These imports bring in everything the generator needs. Faker creates realistic names, emails, and addresses. pyarrow serializes DataFrames to the Parquet columnar format. boto3 uploads the result to S3.

The fixed seeds (42) ensure that every run produces identical data, which is essential for debugging and testing pipelines.

• Add the product generator function below the seed lines:

# Generates 500 products across 8 categories with random pricing
def generate_products(n_products=500):
    categories = [
        "Electronics", "Clothing", "Home & Garden", "Sports",
        "Books", "Toys", "Food & Beverage", "Health & Beauty"
    ]
    brands = [
        "TechPro", "StyleCraft", "HomeBase", "ActiveGear",
        "ReadMore", "FunZone", "FreshChoice", "GlowWell",
        "ValueMax", "PrimeLine"
    ]
    products = []
    for i in range(n_products):
        products.append({
            "product_id": f"PROD-{i+1:05d}",
            "product_name": fake.catch_phrase(),
            "category": random.choice(categories),
            "brand": random.choice(brands),
            "list_price": round(random.uniform(5.99, 599.99), 2),
        })
    return pd.DataFrame(products)

What does this code do?

This function creates 500 product records that will serve as the product dimension seed. Each product gets a unique product_id (like PROD-00001), a category, a brand, and a list price.

The product dimension is small by design. In production, you would pull this from a transactional database rather than generating it.

• Save src/generate_data.py and confirm the imports resolve by running:

python src/generate_data.py

The script exits silently with no output because it only defines functions so far. No error traceback means all imports resolved correctly.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt. If not, run .venv\Scripts\activate (Windows) or source .venv/bin/activate (macOS/Linux) and try again.

help me fix a ModuleNotFoundError when running my generate_data.py script

Add customer and order generation logic

The customer generator creates 10,000 customers with a weighted tier distribution. The order generator is the core of this script. It produces 170,000 orders per day with intentional data quality problems that you will fix in the Spark transformation step.

• Add the customer generator function below generate_products:

# Generates 10,000 customers with weighted tier distribution
def generate_customers(n_customers=10000):
    tiers = ["Bronze", "Silver", "Gold", "Platinum"]
    customers = []
    for i in range(n_customers):
        signup = fake.date_between(start_date="-3y", end_date="-30d")
        customers.append({
            "customer_id": f"CUST-{i+1:06d}",
            "customer_name": fake.name(),
            "email": fake.email(),
            "signup_date": signup.isoformat(),
            # 40% Bronze, 30% Silver, 20% Gold, 10% Platinum
            "tier": random.choices(tiers, weights=[40, 30, 20, 10])[0],
            "city": fake.city(),
            "country": fake.country_code(),
        })
    return pd.DataFrame(customers)

What does this code do?

This creates 10,000 customer records with a realistic tier distribution. Most customers are Bronze (40%), fewer are Platinum (10%). The tier field will change over time, which is why you will track it with SCD Type 2 in Snowflake later.

• Add the order generator function below generate_customers. This first section defines the function and builds each order record:

# Generates orders for one day with intentional duplicates and late arrivals
def generate_orders_for_date(
    processing_date, customer_ids, product_ids,
    orders_per_day=170000, duplicate_rate=0.03, late_arrival_rate=0.05,
):
    payment_methods = ["credit_card", "debit_card", "paypal", "bank_transfer"]
    orders = []

    for _ in range(orders_per_day):
        order_id = str(uuid.uuid4())
        event_date = processing_date

        # 5% of orders are late arrivals from 2-5 days ago
        if random.random() < late_arrival_rate:
            days_late = random.randint(2, 5)
            event_date = processing_date - timedelta(days=days_late)

        order_ts = datetime.combine(event_date, datetime.min.time()) + timedelta(
            hours=random.randint(0, 23),
            minutes=random.randint(0, 59),
            seconds=random.randint(0, 59),
        )

What does this code do?

The function accepts a processing_date (when the data is ingested) and generates 170,000 order records per call. For 5% of orders, the event_date is pushed back 2-5 days to simulate late-arriving data.

The order_ts combines the event date with a random time of day, creating a realistic order timestamp that may differ from the processing date partition.

• Still inside the for loop, add the orders.append block and the duplicate injection logic that closes the function:

        orders.append({
            "order_id": order_id,
            "customer_id": random.choice(customer_ids),
            "product_id": random.choice(product_ids),
            "quantity": random.randint(1, 5),
            "unit_price": round(random.uniform(9.99, 499.99), 2),
            "order_timestamp": order_ts.isoformat(),
            "shipping_address": fake.address().replace("\n", ", "),
            "payment_method": random.choice(payment_methods),
            "processing_date": processing_date.isoformat(),
        })

    # Inject 3% duplicates with slightly offset timestamps to simulate replay
    n_duplicates = int(len(orders) * duplicate_rate)
    for i in range(n_duplicates):
        dup = orders[random.randint(0, len(orders) - 1)].copy()
        dup["order_timestamp"] = (
            datetime.fromisoformat(dup["order_timestamp"])
            + timedelta(seconds=random.randint(1, 300))
        ).isoformat()
        orders.append(dup)

    return pd.DataFrame(orders)

Why inject late arrivals and duplicates?

In production, data rarely arrives perfectly. Orders can be replayed from upstream systems (duplicates with the same order_id but slightly different timestamps). Events can arrive days after they happened (late arrivals).

The 3% duplicate injection picks existing orders at random and appends copies with timestamps offset by 1-300 seconds. The 5% late arrival rate means about 8,500 orders per day have an order_timestamp that is 2-5 days before the processing_date partition they land in.

• Add the S3 upload helper function below generate_orders_for_date:

# Serializes a DataFrame to Parquet in memory and uploads to S3
def upload_parquet_to_s3(df, s3_bucket, s3_key):
    table = pa.Table.from_pandas(df)
    buffer = pa.BufferOutputStream()
    pq.write_table(table, buffer)
    s3 = boto3.client("s3")
    s3.put_object(Bucket=s3_bucket, Key=s3_key, Body=buffer.getvalue().to_pybytes())

What does this code do?

This helper converts a pandas DataFrame to a PyArrow table, writes it as Parquet into an in-memory buffer, then uploads the bytes directly to S3 using put_object. This avoids writing temporary files to disk.

• Save src/generate_data.py and run it again to confirm all functions parse correctly:

python src/generate_data.py

The script still exits silently because there is no __main__ block calling the functions yet. A clean exit confirms the syntax is valid.

Add the orchestrator and run the pipeline

The run function ties everything together. It generates dimension seeds (products, customers), then loops through 30 days creating partitioned order files in S3.

• Add the run function below upload_parquet_to_s3. This first section handles date calculation and dimension uploads:

def run(s3_bucket, target_date=None, days_to_generate=30):
    # Process a single target date or generate 30 days of history
    if target_date:
        dates = [datetime.strptime(target_date, "%Y-%m-%d").date()]
    else:
        end_date = datetime.now().date() - timedelta(days=1)
        dates = [end_date - timedelta(days=i) for i in range(days_to_generate)]

    print("Generating products and customers...")
    products_df = generate_products()
    customers_df = generate_customers()

    product_ids = products_df["product_id"].tolist()
    customer_ids = customers_df["customer_id"].tolist()

    # Upload dimension seeds as single Parquet files
    upload_parquet_to_s3(products_df, s3_bucket, "raw/products/products.parquet")
    upload_parquet_to_s3(customers_df, s3_bucket, "raw/customers/customers.parquet")
    print(f"Uploaded {len(products_df)} products, {len(customers_df)} customers")

What does this code do?

The run function can operate in two modes: process a single target_date (for incremental Airflow runs) or generate a full 30-day backfill (the default).

It generates the product and customer dimension seeds first, uploads them to raw/products/ and raw/customers/, then uses the ID lists to build realistic foreign key references in order records.

• Add the order generation loop that completes the run function, directly below the dimension upload section:

    # Generate and upload orders day by day, partitioned by processing date
    total_orders = 0
    for date in sorted(dates):
        orders_df = generate_orders_for_date(
            processing_date=date, customer_ids=customer_ids, product_ids=product_ids,
        )
        s3_key = (
            f"raw/orders/year={date.year}/month={date.month:02d}/"
            f"day={date.day:02d}/orders_{date.isoformat()}.parquet"
        )
        upload_parquet_to_s3(orders_df, s3_bucket, s3_key)
        total_orders += len(orders_df)
        print(f"  {date}: {len(orders_df)} orders uploaded")

    print(f"Total orders generated: {total_orders:,}")
    return total_orders

Why partition by processing_date instead of order_timestamp?

The raw layer uses processing_date (when the file arrived) as the partition key, not order_timestamp (when the event happened). This is an intentional architectural decision.

Raw data is immutable and append-only. A late-arriving order from 3 days ago still lands in today's partition because that is when it was ingested. The Spark transformation step will re-partition by the correct order_date in the curated layer.

This separation between arrival time and event time is a senior-level data engineering concept frequently asked in interviews.

• Add the entry point at the very bottom of the file:

if __name__ == "__main__":
    import os
    bucket = os.environ.get("S3_BUCKET", "ecommerce-pipeline-dev")
    run(s3_bucket=bucket)

What does this code do?

The __main__ guard ensures run() only executes when you run the script directly. It reads the S3_BUCKET environment variable from your shell (which you set in Step 1) so the script knows which bucket to upload to.

• Save src/generate_data.py.
• Before running, think about what you expect to see. The script will generate 30 days of data at 170,000+ orders per day. How many total orders do you expect?
• Run the data generator by executing this command (this will take several minutes as it creates and uploads millions of rows):

python src/generate_data.py

You should see output showing progress for each day, ending with a total count around 5.25 million orders. The 3% duplicate injection means each day's count is approximately 175,100 (170,000 base + ~5,100 duplicates).

Seeing a NoCredentialsError or AccessDenied?

Make sure your S3_BUCKET environment variable matches the bucket you created in Step 1. Run echo $S3_BUCKET (macOS/Linux) or echo %S3_BUCKET% (Windows) to check. If it is empty, export it before running the script:

export S3_BUCKET=ecommerce-pipeline-your-initials-dev

Also verify your AWS CLI credentials are still valid by running aws s3 ls.

help me fix AWS credential errors when running my generate_data.py script

✔️ Awesome, I've got everything!

Great. Double check that you have saved src/generate_data.py and that your script ran to completion with a total order count in the millions.

ⓧ I'd like to double check the full code

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

• Imports block (uuid, random, datetime, pandas, pyarrow, boto3, faker) with Faker seed setup.
• generate_products function (500 products, 8 categories, 10 brands).
• generate_customers function (10,000 customers, weighted tier distribution).
• generate_orders_for_date function (170,000 orders/day, 5% late arrivals, 3% duplicates).
• upload_parquet_to_s3 helper function (DataFrame to Parquet buffer to S3).
• run orchestrator function (generates dimensions, loops 30 days, uploads partitioned orders).
• if __name__ == "__main__" entry point reading S3_BUCKET from environment.

Verify the data lake structure in S3

The raw layer should now contain immutable, append-only Parquet files organized by processing date. The partition scheme follows the year=YYYY/month=MM/day=DD/ Hive-style convention that both Spark and Snowflake understand natively.

• Before running the next command, predict what the S3 path structure will look like. Will the partitions use the order_timestamp or the processing_date?
• List the order partitions in S3 by running:

aws s3 ls s3://[[S3_BUCKET="ecommerce-pipeline-your-initials-dev"]]/raw/orders/ --recursive | head -40

You should see Parquet files organized by year=YYYY/month=MM/day=DD/ partitions, with one file per day. There should be 30 partition folders covering the last 30 days.

• Confirm the dimension seed files exist by running:

aws s3 ls s3://[[S3_BUCKET="ecommerce-pipeline-your-initials-dev"]]/raw/products/
aws s3 ls s3://[[S3_BUCKET="ecommerce-pipeline-your-initials-dev"]]/raw/customers/

You should see products.parquet and customers.parquet listed.

What is the data lake structure?

Your S3 bucket now holds the raw layer of a data lake with three prefixes:

• raw/products/ contains a single Parquet file with 500 product records.
• raw/customers/ contains a single Parquet file with 10,000 customer records.
• raw/orders/year=.../month=.../day=.../ contains one Parquet file per processing day with ~175,100 rows each.

The raw layer is immutable. You never modify or delete files here. Every downstream transformation reads from this layer and writes to a separate curated layer.

S3 listing is empty?

Check that your script printed progress lines for each day. If it exited without output, the S3_BUCKET variable might not be set. Also double-check the bucket name in your aws s3 ls command matches the one you created.

help me troubleshoot why my S3 bucket is empty after running generate_data.py

Your data lake raw layer is loaded with 5M+ realistic order events, complete with the real-world data quality problems that make batch pipelines interesting. Next up, you will build the PySpark transformation that cleans and enriches this data into a curated layer.

Build the PySpark Transformation

Your raw data layer now holds over 5 million order records in S3. The problem is that this data is messy. It contains 3% duplicate orders (simulating message replays), 5% late-arriving records sitting in the wrong date partition, and potentially null values in required fields.

You need a transformation engine that can handle millions of rows with joins against dimension tables without running out of memory. PySpark provides exactly that. In this step, you will build a Spark job that deduplicates, validates, enriches, and re-partitions the raw data into a clean curated layer.

In this step, get ready to:

• Deduplicate raw orders using window functions.
• Enrich orders with product dimensions and derived columns.
• Run the transformation via Docker and verify the curated output.

Create the Spark session and cleaning functions

The first piece of the transformation sets up how Spark connects to S3 and how it handles partitions on re-write. The deduplication logic uses a window function that is the standard pattern for keeping one record per key in distributed data.

• Create a new file called src/transform.py in your project folder, and add the following imports and Spark session factory:

import os
import uuid
from datetime import datetime

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window


def create_spark_session():
    return (
        SparkSession.builder
        .appName("EcommerceOrderTransform")
        .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
        .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
        .config("spark.hadoop.fs.s3a.aws.credentials.provider",
                "com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
        .getOrCreate()
    )

What does this code do?

The partitionOverwriteMode set to dynamic is what makes this job idempotent. When you re-run it for the same date, Spark only overwrites the partitions it touches instead of wiping the entire output directory.

The S3A configuration tells Spark to use the Hadoop S3A filesystem connector and pick up AWS credentials from the environment automatically.

• Below create_spark_session, add the deduplication and validation functions:

def deduplicate_orders(df):
    # Keep only the earliest record per order_id
    window = Window.partitionBy("order_id").orderBy(F.col("order_timestamp").asc())
    deduped = (
        df.withColumn("row_num", F.row_number().over(window))
        .filter(F.col("row_num") == 1)
        .drop("row_num")
    )
    return deduped


def validate_and_clean(df):
    # Drop rows missing required identifiers
    valid = df.filter(F.col("order_id").isNotNull() & F.col("customer_id").isNotNull())
    dropped_count = df.count() - valid.count()
    if dropped_count > 0:
        print(f"WARNING: Dropped {dropped_count} rows with null order_id or customer_id")
    return valid

What does this code do?

The deduplicate_orders function uses ROW_NUMBER() partitioned by order_id and ordered by order_timestamp ascending. This keeps the earliest record when duplicates share the same order_id.

The validate_and_clean function enforces non-null constraints on order_id and customer_id. Any dropped rows get logged so you can investigate upstream data quality issues.

• Save src/transform.py.
• Build the Docker image so the Spark runtime is ready when you finish the file. Run this command:

docker compose build

You will see Docker downloading the base Airflow image, installing Java 17, and pip-installing PySpark. This takes 2-5 minutes on first build.

Build failing with network errors?

Make sure Docker Desktop is running. If you see timeout errors, check your internet connection and retry. Docker caches layers, so a second attempt is faster.

help me debug my Docker build failure

Add enrichment and metadata functions

The enrichment step joins orders with product data to add category, brand, and list price. It also computes derived columns that analysts need and flags late-arriving records so the curated layer partitions them correctly.

• In src/transform.py, add the enrichment and metadata functions below validate_and_clean:

def enrich_orders(orders_df, products_df):
    # Join product dimension and compute derived columns
    enriched = (
        orders_df
        .join(products_df, on="product_id", how="left")
        .withColumn("total_amount", F.round(F.col("quantity") * F.col("unit_price"), 2))
        .withColumn("order_date", F.to_date(F.col("order_timestamp")))
        .withColumn("is_late_arrival",
                    F.col("order_date") != F.to_date(F.col("processing_date")))
    )
    return enriched


def add_metadata(df, pipeline_run_id):
    # Tag every row with lineage tracking columns
    return (
        df
        .withColumn("_pipeline_run_id", F.lit(pipeline_run_id))
        .withColumn("_processed_at", F.lit(datetime.utcnow().isoformat()))
    )

Why re-partition by order_date?

The raw layer partitions by processing_date (when the file arrived in S3). A late-arriving order from Monday that lands on Wednesday sits in Wednesday's raw partition.

The order_date column extracts the actual event date from order_timestamp. The curated layer partitions by this column, so analysts querying Monday's orders find all of them in one place regardless of when they arrived.

The is_late_arrival flag lets you track how many records needed re-partitioning. This is a key data quality metric in production.

• Now add the run function that orchestrates all the helpers. Paste this below add_metadata:

def run(s3_bucket, processing_date):
    spark = create_spark_session()
    pipeline_run_id = str(uuid.uuid4())

    print(f"Processing date: {processing_date}")
    print(f"Pipeline run ID: {pipeline_run_id}")

    date_obj = datetime.strptime(processing_date, "%Y-%m-%d").date()
    raw_path = (
        f"s3a://{s3_bucket}/raw/orders/"
        f"year={date_obj.year}/month={date_obj.month:02d}/day={date_obj.day:02d}/"
    )
    products_path = f"s3a://{s3_bucket}/raw/products/products.parquet"
    output_path = f"s3a://{s3_bucket}/curated/orders/"

    print(f"Reading raw orders from: {raw_path}")
    raw_orders = spark.read.parquet(raw_path)
    raw_count = raw_orders.count()
    print(f"Raw order count: {raw_count:,}")

    products = spark.read.parquet(products_path)

    deduped = deduplicate_orders(raw_orders)
    deduped_count = deduped.count()
    print(f"After deduplication: {deduped_count:,} (removed {raw_count - deduped_count:,})")

    cleaned = validate_and_clean(deduped)
    enriched = enrich_orders(cleaned, products)
    final = add_metadata(enriched, pipeline_run_id)

What does this code do?

The run function is the entry point. It builds the S3 path for the target date, reads the raw Parquet data, then pipes it through deduplication, validation, enrichment, and metadata tagging in sequence.

Notice the path uses s3a:// (not s3://). The S3A connector is Spark's preferred interface to Amazon S3 and supports the Hadoop credential chain configured in the session.

• Verify your file has no syntax errors so far by running this locally:

python -c "import ast; ast.parse(open('src/transform.py').read()); print('Syntax valid')"

You should see Syntax valid printed. This confirms all your functions parse correctly before you finish the file.

Seeing a SyntaxError?

Check for mismatched parentheses in the run function. Each f-string with :, formatting must have matching braces. Also verify you have consistent 4-space indentation throughout.

help me find the syntax error in my transform.py

Complete the output logic and run the job

The final section of run() selects the columns for the curated output, writes partitioned Parquet to S3, and reports the results. The main block at the bottom makes the script runnable from the command line.

• Add this code at the bottom of src/transform.py, continuing inside the run function (same indentation level as the final = line above):

    output_cols = [
        "order_id", "customer_id", "product_id", "product_name",
        "category", "brand", "quantity", "unit_price", "list_price",
        "total_amount", "order_timestamp", "order_date",
        "shipping_address", "payment_method",
        "is_late_arrival", "processing_date",
        "_pipeline_run_id", "_processed_at",
    ]
    final_output = final.select(*output_cols)

    print("Writing curated data partitioned by order_date...")
    final_output.write.mode("overwrite").partitionBy("order_date").parquet(output_path)

    final_count = final_output.count()
    late_count = final_output.filter(F.col("is_late_arrival") == True).count()
    print(f"Curated output: {final_count:,} rows, {late_count:,} late arrivals")

    spark.stop()
    return {"rows_processed": final_count, "duplicates_removed": raw_count - deduped_count}


if __name__ == "__main__":
    bucket = os.environ.get("S3_BUCKET", "ecommerce-pipeline-dev")
    date = os.environ.get("PROCESSING_DATE", "2026-07-10")
    run(s3_bucket=bucket, processing_date=date)

What does this code do?

The output_cols list explicitly selects which columns make it to the curated layer. This prevents internal columns (like row_num) from leaking into the output.

The .write.mode("overwrite").partitionBy("order_date") line uses dynamic partition overwrite. Only the partitions that this run touches get replaced. Other dates remain untouched. This is what makes re-running safe.

The _pipeline_run_id and _processed_at metadata columns provide lineage tracking. You can trace any curated record back to the exact pipeline execution that produced it.

• Save src/transform.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before running it.

ⓧ I'd like to double check the full code

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

• Imports (os, uuid, datetime, SparkSession, functions as F, Window).
• create_spark_session() with dynamic partition overwrite and S3A config.
• deduplicate_orders(df) using ROW_NUMBER window function.
• validate_and_clean(df) filtering nulls with a warning log.
• enrich_orders(orders_df, products_df) with join, total_amount, order_date, is_late_arrival.
• add_metadata(df, pipeline_run_id) adding _pipeline_run_id and _processed_at.
• run(s3_bucket, processing_date) orchestrating the full pipeline with output column selection.
• if __name__ == "__main__" block reading S3_BUCKET and PROCESSING_DATE from environment.

Before you run the transformation, think about this: given that the raw data for one day has ~175,000 rows (170,000 orders plus 3% duplicates), do you expect the curated output to have more rows, fewer rows, or the same number?

• Run the PySpark transformation for a single day via Docker by running this command:

docker compose run --rm --no-deps -e PROCESSING_DATE=2026-07-10 airflow-webserver python /opt/airflow/src/transform.py

What does this command do?

The docker compose run command starts a one-off container from the airflow-webserver service definition (which has Java 17 and PySpark installed). The --rm flag deletes the container after it finishes. The --no-deps flag prevents Docker from starting Postgres, since you only need the Spark runtime.

You should see output showing the raw count, the count after deduplication (fewer rows, confirming duplicates were removed), and the final curated row count with late arrival stats.

• Verify the curated data landed in S3 by running:

aws s3 ls s3://[[S3_BUCKET="ecommerce-pipeline-yourname-dev"]]/curated/orders/ --recursive | head -20

You should see Parquet files organized under order_date=YYYY-MM-DD/ partitions. Late-arriving orders appear in their correct event date partition, not the processing date they arrived on.

Not seeing curated files in S3?

Check the Docker output for errors. Common issues: wrong S3_BUCKET value in your .env file, or AWS credentials that have expired. Verify your bucket name matches exactly by running aws s3 ls to list all your buckets.

If you see a Java or Spark error, make sure docker compose build completed successfully earlier. The image must include Java 17 for PySpark to work.

help me debug why my PySpark transform is not writing to S3

Your raw data is now clean, deduplicated, and enriched in the curated layer. Next up, you will build the Snowflake dimensional model that transforms this curated data into a star schema analysts can query directly.

Build the Snowflake Dimensional Model

Your curated Parquet layer is clean and deduplicated, but analysts cannot query Parquet files directly from their BI tools. They need a star schema with proper dimension tables, surrogate keys, and historical tracking.

Snowflake provides the query engine, concurrency, and BI tool connectivity that makes this possible. In this step, you will create a dimensional model (fact table + dimensions) in Snowflake and build a Python loader that uses COPY INTO to bulk-load data from S3.

In this step, get ready to:

• Create the Snowflake dimensional model DDL with staging and analytics schemas.
• Build a Python loader that populates the star schema from S3.
• Verify the load is idempotent and data appears in the warehouse.

Create the dimensional model DDL

The dimensional model lives in a SQL setup script that creates everything Snowflake needs: a database, schemas for staging and analytics, an external stage pointing at your S3 bucket, and the star schema tables.

• Create a snowflake/ directory in your project folder by running:

mkdir snowflake

• Create a new file called setup.sql inside the snowflake/ directory.
• Add the database, schemas, file format, storage integration, and external stage by pasting this code:

CREATE DATABASE IF NOT EXISTS ECOMMERCE_DW;
USE DATABASE ECOMMERCE_DW;

CREATE SCHEMA IF NOT EXISTS staging;
CREATE SCHEMA IF NOT EXISTS analytics;

CREATE OR REPLACE FILE FORMAT staging.parquet_format
  TYPE = 'PARQUET'
  COMPRESSION = 'SNAPPY';

CREATE OR REPLACE STORAGE INTEGRATION s3_integration
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::[[AWS_ACCOUNT_ID="your-account-id"]]:role/snowflake-access-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://[[S3_BUCKET="ecommerce-pipeline-{initials}-dev"]]/');

CREATE OR REPLACE STAGE staging.curated_stage
  STORAGE_INTEGRATION = s3_integration
  URL = 's3://[[S3_BUCKET="ecommerce-pipeline-{initials}-dev"]]/curated/orders/'
  FILE_FORMAT = staging.parquet_format;

What does this code do?

• Creates the ECOMMERCE_DW database with two schemas: staging (for temporary landing) and analytics (for the final star schema).
• The storage integration authenticates Snowflake to your S3 bucket using an IAM role.
• The external stage curated_stage points directly at your curated orders prefix so COPY INTO can pull Parquet files.

How do I set up the IAM role for Snowflake?

The storage integration requires an IAM role in AWS that Snowflake can assume. Before running this SQL, you need to:

• Create an IAM role in AWS with a trust policy that allows Snowflake to assume it.
• Attach a policy granting s3:GetBucketLocation, s3:GetObject, s3:GetObjectVersion, and s3:ListBucket on your bucket.
• After creating the integration in Snowflake, run DESC INTEGRATION s3_integration; to get the STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID values.
• Update your IAM role's trust policy with those values so Snowflake can assume it.

• Copy the infrastructure section (everything you just added) into a Snowflake worksheet and run it.
• Verify the schemas were created by running:

SHOW SCHEMAS IN DATABASE ECOMMERCE_DW;

You should see STAGING and ANALYTICS listed alongside the default PUBLIC and INFORMATION_SCHEMA schemas.

• Next, add the staging tables to setup.sql below the stage creation:

CREATE OR REPLACE TRANSIENT TABLE staging.stg_orders (
  order_id VARCHAR(36),
  customer_id VARCHAR(11),
  product_id VARCHAR(10),
  product_name VARCHAR(200),
  category VARCHAR(50),
  brand VARCHAR(50),
  quantity INTEGER,
  unit_price DECIMAL(10,2),
  list_price DECIMAL(10,2),
  total_amount DECIMAL(12,2),
  order_timestamp TIMESTAMP_NTZ,
  order_date DATE,
  shipping_address VARCHAR(500),
  payment_method VARCHAR(20),
  is_late_arrival BOOLEAN,
  processing_date DATE,
  pipeline_run_id VARCHAR(36),
  processed_at TIMESTAMP_NTZ
);

CREATE OR REPLACE TRANSIENT TABLE staging.stg_customers (
  customer_id VARCHAR(11),
  customer_name VARCHAR(200),
  email VARCHAR(200),
  signup_date DATE,
  tier VARCHAR(20),
  city VARCHAR(100),
  country VARCHAR(5)
);

Why are staging tables transient?

Transient tables skip Snowflake's Time Travel and Fail-safe storage. Staging data is temporary and gets truncated before each load, so paying for recovery features on it is wasteful.

• Now add the date dimension table with its pre-populated data below the staging tables:

CREATE OR REPLACE TABLE analytics.dim_dates (
  date_key INTEGER PRIMARY KEY,
  full_date DATE,
  day_of_week INTEGER,
  day_name VARCHAR(10),
  month INTEGER,
  month_name VARCHAR(10),
  quarter INTEGER,
  year INTEGER,
  is_weekend BOOLEAN
);

INSERT INTO analytics.dim_dates
SELECT
  TO_NUMBER(TO_CHAR(d.date_val, 'YYYYMMDD')) AS date_key,
  d.date_val AS full_date,
  DAYOFWEEK(d.date_val) AS day_of_week,
  DAYNAME(d.date_val) AS day_name,
  MONTH(d.date_val) AS month,
  MONTHNAME(d.date_val) AS month_name,
  QUARTER(d.date_val) AS quarter,
  YEAR(d.date_val) AS year,
  CASE WHEN DAYOFWEEK(d.date_val) IN (0, 6) THEN TRUE ELSE FALSE END AS is_weekend
FROM (
  SELECT DATEADD(day, SEQ4(), '2020-01-01')::DATE AS date_val
  FROM TABLE(GENERATOR(ROWCOUNT => 4018))
) d;

What does the GENERATOR do?

Snowflake's GENERATOR function produces 4018 rows (covering 2020-01-01 through 2030-12-31). Each row becomes a calendar date with derived attributes like day name, quarter, and weekend flag.

Pre-populating the date dimension means every order_date in your fact table can always find a matching date_key without runtime generation.

• Paste the staging tables and dim_dates SQL into your Snowflake worksheet and run them.
• Verify the date dimension populated correctly by running:

SELECT COUNT(*) FROM analytics.dim_dates;

You should see 4018 rows, covering every calendar date from 2020 through 2030.

• Add the product and customer dimension tables to setup.sql:

CREATE OR REPLACE TABLE analytics.dim_products (
  product_key INTEGER AUTOINCREMENT PRIMARY KEY,
  product_id VARCHAR(10),
  product_name VARCHAR(200),
  category VARCHAR(50),
  brand VARCHAR(50),
  list_price DECIMAL(10,2),
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE OR REPLACE TABLE analytics.dim_customers (
  customer_key INTEGER AUTOINCREMENT PRIMARY KEY,
  customer_id VARCHAR(11),
  customer_name VARCHAR(200),
  email VARCHAR(200),
  tier VARCHAR(20),
  city VARCHAR(100),
  country VARCHAR(5),
  valid_from DATE,
  valid_to DATE DEFAULT '9999-12-31',
  is_current BOOLEAN DEFAULT TRUE,
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

What is SCD Type 2?

Slowly Changing Dimension Type 2 tracks historical changes. When a customer's tier changes from Silver to Gold, the old record gets expired (valid_to set to today, is_current set to FALSE) and a new record is inserted with the updated tier.

This lets analysts answer "what tier was this customer when they placed that order?" rather than only knowing their current tier.

• Finally, add the fact table at the bottom of setup.sql:

CREATE OR REPLACE TABLE analytics.fact_orders (
  order_key INTEGER AUTOINCREMENT PRIMARY KEY,
  order_id VARCHAR(36),
  customer_key INTEGER REFERENCES analytics.dim_customers(customer_key),
  product_key INTEGER REFERENCES analytics.dim_products(product_key),
  date_key INTEGER REFERENCES analytics.dim_dates(date_key),
  quantity INTEGER,
  unit_price DECIMAL(10,2),
  total_amount DECIMAL(12,2),
  payment_method VARCHAR(20),
  is_late_arrival BOOLEAN,
  pipeline_run_id VARCHAR(36),
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

Why surrogate keys and foreign keys?

The fact_orders table uses surrogate keys (customer_key, product_key, date_key) instead of natural keys. This connects each order to the correct historical version of a dimension record.

The pipeline_run_id column tracks which pipeline execution loaded each row. This is the key to making the load idempotent.

• Paste the dimension and fact table SQL into your Snowflake worksheet and run it.
• Confirm all analytics tables exist by running:

SHOW TABLES IN SCHEMA analytics;

You should see four tables listed: DIM_CUSTOMERS, DIM_DATES, DIM_PRODUCTS, and FACT_ORDERS.

• Save snowflake/setup.sql in your editor.

Storage integration failing?

The CREATE STORAGE INTEGRATION command requires the ACCOUNTADMIN role or a role with the CREATE INTEGRATION privilege. Run USE ROLE ACCOUNTADMIN; before executing the integration statement.

If you see a trust policy error, make sure you completed the IAM role setup and updated the trust policy with the values from DESC INTEGRATION s3_integration;.

help me fix my Snowflake storage integration for S3

✔️ Awesome, I've got everything!

Great. Make sure snowflake/setup.sql is saved. The full file contains all five sections you just ran.

ⓧ I'd like to double check the full code

Your snowflake/setup.sql should contain these five sections in order. Scroll up and confirm each is present:

• Database, schemas, file format, storage integration, and external stage.
• Transient staging tables (stg_orders and stg_customers).
• The dim_dates table with its INSERT using GENERATOR.
• The dim_products and dim_customers tables (dim_customers has SCD Type 2 columns).
• The fact_orders table with FK references to all three dimensions.

Build the Python loader

The dimensional model exists, but it is empty. You need a Python script that pulls curated Parquet from S3 into staging, applies SCD Type 2 logic to the customer dimension, and inserts facts with proper dimension key lookups.

• Create a new file src/snowflake_loader.py and add the connection helper:

import os
from datetime import datetime

import snowflake.connector


def get_connection():
    # Connect using environment variables from .env
    return snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
        database="ECOMMERCE_DW",
        warehouse="PIPELINE_WH",
    )

What does this code do?

The get_connection() function reads your Snowflake credentials from environment variables (set in your .env file) and returns an authenticated connection pointing at the ECOMMERCE_DW database and PIPELINE_WH warehouse.

• Verify your connection works by running a quick test:

python -c "from src.snowflake_loader import get_connection; c = get_connection(); print('Connected:', not c.is_closed()); c.close()"

You should see Connected: True printed. This confirms your Snowflake credentials are correct.

Connection refused or authentication error?

Double-check your .env file has the correct SNOWFLAKE_ACCOUNT (the identifier from your Snowflake URL, not the full URL), SNOWFLAKE_USER, and SNOWFLAKE_PASSWORD. Make sure your virtual environment is activated so the environment variables are loaded.

help me debug my Snowflake connection in Python

• Add the staging load function below get_connection():

def load_staging(conn, processing_date):
    cursor = conn.cursor()
    # Truncate before loading to ensure idempotency
    cursor.execute("TRUNCATE TABLE staging.stg_orders;")

    # COPY INTO pulls Parquet from S3 via the external stage
    copy_sql = """
    COPY INTO staging.stg_orders
    FROM @staging.curated_stage
    PATTERN = '.*order_date={}.*'
    FILE_FORMAT = staging.parquet_format
    MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;
    """.format(processing_date)

    cursor.execute(copy_sql)
    result = cursor.fetchone()
    row_count = result[0] if result else 0
    print(f"Loaded {row_count} rows into staging.stg_orders")
    cursor.close()
    return row_count

What does this code do?

• TRUNCATE ensures staging is empty before loading. If the pipeline retries, you get a clean slate.
• The PATTERN parameter filters S3 files to only load the target date's partition from the curated stage.
• MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE maps Parquet column names to table columns regardless of case differences.

• Add the first part of the SCD Type 2 merge function. This MERGE statement expires stale records and inserts brand-new customers:

def merge_dim_customers(conn):
    cursor = conn.cursor()
    # MERGE handles two cases: expire changed records, insert new customers
    merge_sql = """
    MERGE INTO analytics.dim_customers AS target
    USING (
        SELECT DISTINCT customer_id, customer_name, email, tier, city, country
        FROM staging.stg_customers
    ) AS source
    ON target.customer_id = source.customer_id AND target.is_current = TRUE
    WHEN MATCHED AND (target.tier != source.tier OR target.email != source.email)
    THEN UPDATE SET valid_to = CURRENT_DATE(), is_current = FALSE
    WHEN NOT MATCHED THEN INSERT (
        customer_id, customer_name, email, tier, city, country, valid_from, valid_to, is_current
    ) VALUES (
        source.customer_id, source.customer_name, source.email,
        source.tier, source.city, source.country,
        CURRENT_DATE(), '9999-12-31', TRUE
    );
    """
    cursor.execute(merge_sql)

How does MERGE implement SCD Type 2?

When a customer's tier or email changed, the MERGE expires the old record (sets valid_to to today and is_current to FALSE). Brand-new customers that have never been seen get inserted directly as current records.

• Still inside merge_dim_customers, add the second operation that inserts new active versions for customers whose records were just expired:

    # Insert new active versions for customers whose old record was just expired
    insert_new_versions = """
    INSERT INTO analytics.dim_customers (
        customer_id, customer_name, email, tier, city, country, valid_from, valid_to, is_current
    )
    SELECT s.customer_id, s.customer_name, s.email, s.tier, s.city, s.country,
           CURRENT_DATE(), '9999-12-31', TRUE
    FROM staging.stg_customers s
    INNER JOIN analytics.dim_customers d
        ON s.customer_id = d.customer_id AND d.is_current = FALSE AND d.valid_to = CURRENT_DATE()
    WHERE NOT EXISTS (
        SELECT 1 FROM analytics.dim_customers e
        WHERE e.customer_id = s.customer_id AND e.is_current = TRUE
    );
    """
    cursor.execute(insert_new_versions)
    cursor.close()

Why a separate INSERT after the MERGE?

The MERGE expired the old record, but Snowflake's MERGE cannot insert a new row for the same customer_id in the same statement. This follow-up INSERT creates the fresh active record with the updated attributes and valid_from = today.

• Add the fact table load function below merge_dim_customers:

def load_fact_orders(conn, pipeline_run_id):
    cursor = conn.cursor()
    # Delete any existing rows from a previous run with the same ID
    cursor.execute(f"DELETE FROM analytics.fact_orders WHERE pipeline_run_id = '{pipeline_run_id}';")

    # Insert facts with dimension key lookups
    insert_sql = """
    INSERT INTO analytics.fact_orders (
        order_id, customer_key, product_key, date_key,
        quantity, unit_price, total_amount, payment_method,
        is_late_arrival, pipeline_run_id
    )
    SELECT
        s.order_id, dc.customer_key, dp.product_key, dd.date_key,
        s.quantity, s.unit_price, s.total_amount, s.payment_method,
        s.is_late_arrival, s.pipeline_run_id
    FROM staging.stg_orders s
    LEFT JOIN analytics.dim_customers dc
        ON s.customer_id = dc.customer_id AND dc.is_current = TRUE
    LEFT JOIN analytics.dim_products dp ON s.product_id = dp.product_id
    LEFT JOIN analytics.dim_dates dd ON s.order_date = dd.full_date;
    """
    cursor.execute(insert_sql)
    row_count = cursor.rowcount
    print(f"Inserted {row_count} rows into analytics.fact_orders")
    cursor.close()
    return row_count

What does this code do?

• The DELETE-then-INSERT pattern keyed on pipeline_run_id makes this idempotent. Re-running for the same pipeline execution replaces the previous rows instead of duplicating them.
• The SELECT joins staging orders with all three dimension tables to resolve surrogate keys (customer_key, product_key, date_key).
• LEFT JOINs ensure orders still load even if a dimension record is missing. The key will be NULL, which is flagged during quality checks later.

• Add the orchestration function and main block at the bottom of the file:

def run(processing_date, pipeline_run_id=None):
    conn = get_connection()
    try:
        row_count = load_staging(conn, processing_date)
        if row_count == 0:
            print(f"No data for {processing_date}, skipping.")
            return
        merge_dim_customers(conn)
        load_fact_orders(conn, pipeline_run_id or "manual-run")
        print(f"Successfully loaded data for {processing_date}")
    finally:
        conn.close()


if __name__ == "__main__":
    date = os.environ.get("PROCESSING_DATE", "2026-07-10")
    run(processing_date=date)

What does this code do?

• The run() function orchestrates the full load sequence: stage from S3, merge customers, load facts.
• If staging returns zero rows (no data for that date), it exits early rather than running empty merges.
• The try/finally block ensures the Snowflake connection closes even if an error occurs mid-load.

• Save src/snowflake_loader.py.

✔️ Awesome, I've got everything!

Great. Double check you saved the file before moving on to the verification.

ⓧ I'd like to double check the full code

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

• Imports (os, datetime, snowflake.connector) and get_connection().
• load_staging() with TRUNCATE + COPY INTO.
• merge_dim_customers() with MERGE + INSERT new versions.
• load_fact_orders() with DELETE-then-INSERT.
• run() orchestrator and __main__ block.

Run the loader and verify idempotency

The loader is designed to be re-runnable without creating duplicates. Staging gets truncated before each COPY, and fact_orders uses DELETE-then-INSERT keyed on pipeline_run_id. Time to prove it works.

• Before you run the loader, predict: will the load succeed on the first try, or will it fail? Why?
• Run the loader for a date that has curated data in S3:

python -m src.snowflake_loader

You should see output showing rows loaded into staging, and rows inserted into analytics.fact_orders. The exact count depends on how much curated data exists for the target date.

• Switch to your Snowflake worksheet and verify the fact table has data by running:

SELECT COUNT(*) FROM analytics.fact_orders;

You should see a non-zero count matching the number reported by the loader script.

• Verify the SCD Type 2 customer dimension is populated by running:

SELECT * FROM analytics.dim_customers WHERE is_current = TRUE LIMIT 10;

You should see customer records with is_current = TRUE, valid_from set to today's date, and valid_to set to 9999-12-31.

• Now prove idempotency. Run the loader a second time:

python -m src.snowflake_loader

• Query the count again in Snowflake:

SELECT COUNT(*) FROM analytics.fact_orders;

The count should be identical to the first run. No duplicates were created because the DELETE-then-INSERT pattern removed the previous load before inserting again.

Row count doubled after the second run?

Check that load_fact_orders includes the DELETE statement before the INSERT. Without it, each run appends instead of replacing. Also confirm the pipeline_run_id value is consistent between runs (the default is manual-run).

help me fix duplicate rows in my Snowflake fact table after re-running the loader

Your star schema is live and loaded with real pipeline data. Next up, you will orchestrate the entire pipeline with Dockerized Airflow so it runs end-to-end with a single command.

Orchestrate with Dockerized Airflow

Your Snowflake star schema is built and your loader script can populate it. But right now, running the pipeline means manually executing three Python scripts in the correct order, remembering the right date arguments, and hoping nothing fails midway.

In this step, you wire everything together with Apache Airflow, an orchestration platform that handles scheduling, dependency management, retries, and operational visibility. Running Airflow in Docker mirrors production deployments and costs nothing beyond local compute.

In this step, get ready to:

• Create an Airflow DAG that chains data generation, Spark transformation, Snowflake loading, and staging cleanup.
• Build and launch the Dockerized Airflow stack with a single command.
• Trigger the DAG and verify end-to-end data flow into Snowflake.

Create the Airflow DAG file

An Airflow DAG (Directed Acyclic Graph) defines tasks and the order they run in. Each task in your DAG calls one of the Python scripts you already built, passing the execution date so the pipeline processes one day at a time.

• Create the dags directory inside your project folder by running this command:

mkdir dags

• Create a new file called ecommerce_batch_pipeline.py inside the dags/ folder and add the following imports, configuration, and first task function:

import os
import sys
from datetime import datetime, timedelta

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

# Allow imports from the src directory mounted into the container
sys.path.insert(0, "/opt/airflow/src")

# Retry and dependency settings applied to every task
default_args = {
    "owner": "data-engineering",
    "depends_on_past": True,
    "email_on_failure": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=15),
}


# First task: generate raw order data for the execution date
def task_generate_data(**context):
    from generate_data import run as generate_run
    execution_date = context["ds"]
    s3_bucket = os.environ["S3_BUCKET"]
    generate_run(s3_bucket=s3_bucket, target_date=execution_date)

What does this code do?

The sys.path.insert line tells Python where to find your src/ modules when running inside the Airflow container.

The default_args dictionary applies to every task in the DAG. depends_on_past=True ensures dates are processed in order. retries=3 with exponential backoff means transient failures (network blips, S3 throttling) are retried automatically.

The task_generate_data function pulls the execution date from Airflow's context (context["ds"]) and passes it to your data generator. This makes each run process exactly one day.

• Save dags/ecommerce_batch_pipeline.py.
• Confirm the file exists by running:

ls dags/

You should see ecommerce_batch_pipeline.py listed.

Don't see the file?

Make sure you created the dags/ directory in the same folder that contains your docker-compose.yml. The volume mount expects ./dags relative to the project root.

help me find where my DAG file should go relative to docker-compose.yml

• Add the remaining three task functions below task_generate_data in dags/ecommerce_batch_pipeline.py:

# Second task: deduplicate and enrich with PySpark
def task_spark_transform(**context):
    from transform import run as transform_run
    execution_date = context["ds"]
    s3_bucket = os.environ["S3_BUCKET"]
    result = transform_run(s3_bucket=s3_bucket, processing_date=execution_date)
    context["ti"].xcom_push(key="transform_result", value=result)


# Third task: load curated data into Snowflake star schema
def task_load_snowflake(**context):
    from snowflake_loader import run as loader_run
    execution_date = context["ds"]
    run_id = context["run_id"]
    loader_run(processing_date=execution_date, pipeline_run_id=run_id)


# Fourth task: clean up staging tables after successful load
def task_cleanup_staging(**context):
    from snowflake_loader import get_connection
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("TRUNCATE TABLE staging.stg_orders;")
    cursor.close()
    conn.close()

What do these task functions do?

Each function wraps one of your existing scripts. task_spark_transform also pushes the transform result to XCom so you can inspect row counts in the Airflow UI later.

The task_load_snowflake function passes Airflow's unique run_id as the pipeline_run_id. This is what makes the Snowflake load idempotent. If the task retries, the DELETE-then-INSERT keyed on that ID replaces the previous attempt.

The task_cleanup_staging function truncates the staging table after a successful load. Staging data is transient and should not persist between runs.

• Add the DAG definition and first two task operators at the bottom of dags/ecommerce_batch_pipeline.py:

with DAG(
    dag_id="ecommerce_batch_pipeline",
    default_args=default_args,
    description="E-commerce order batch pipeline: S3 -> Spark -> Snowflake",
    schedule="@daily",
    start_date=datetime(2026, 6, 10),
    catchup=True,
    max_active_runs=1,
    tags=["ecommerce", "batch", "production"],
) as dag:

    generate = PythonOperator(
        task_id="generate_data",
        python_callable=task_generate_data,
    )

    transform = PythonOperator(
        task_id="spark_transform",
        python_callable=task_spark_transform,
    )

What does the DAG configuration do?

The schedule="@daily" tells Airflow to run once per day. Combined with catchup=True and a start_date of June 10, 2026, Airflow will backfill all missed dates when triggered.

The max_active_runs=1 setting ensures only one backfill run executes at a time. This prevents parallel runs from competing for Snowflake staging tables.

• Still inside the with DAG(...) as dag: block, add the remaining two operators and the dependency chain below the transform operator:

    load = PythonOperator(
        task_id="load_snowflake",
        python_callable=task_load_snowflake,
    )

    cleanup = PythonOperator(
        task_id="cleanup_staging",
        python_callable=task_cleanup_staging,
    )

    # Define execution order: generate -> transform -> load -> cleanup
    generate >> transform >> load >> cleanup

What does the chain operator do?

The >> operator sets the execution order: generate data first, then transform, then load Snowflake, then clean up staging. If any upstream task fails, downstream tasks do not run.

• Save dags/ecommerce_batch_pipeline.py.
• Build the custom Airflow Docker image (which installs Java 17 and your Python dependencies) by running:

docker compose build

You should see Docker pulling the base Airflow image, installing Java, and installing packages from requirements-airflow.txt. This takes 2-5 minutes on the first build.

Build failing?

If you see a network error during apt-get update, check that Docker Desktop is running and has internet access.

If pip install fails with a memory error, increase Docker's RAM allocation to at least 4 GB in Docker Desktop under Settings then Resources.

help me debug my Docker Compose build failure for the Airflow image

✔️ Awesome, I've got everything!

Great. Double check you have saved dags/ecommerce_batch_pipeline.py before moving on.

ⓧ I'd like to double check the full code

Your dags/ecommerce_batch_pipeline.py file is over 30 lines, so confirm each section is present in order:

• Chunk 1: Imports (os, sys, datetime, DAG, PythonOperator), sys.path.insert, default_args dictionary, and task_generate_data function.
• Chunk 2: task_spark_transform, task_load_snowflake, and task_cleanup_staging functions.
• Chunk 3: with DAG(...) as dag: block with four PythonOperator tasks and the generate >> transform >> load >> cleanup chain.

Launch the Airflow stack

Your docker-compose.yml from earlier defines the full Airflow architecture: a Postgres metadata database, an initialization service that migrates the DB and creates an admin user, a webserver, and a scheduler.

• Start all services in detached mode by running:

docker compose up -d

What does this command do?

Docker Compose starts Postgres first (since other services depend on it), then runs airflow-init to migrate the database and create the admin user (airflow/airflow). Once the init container exits with code 0, the webserver and scheduler start.

The -d flag runs everything in the background so you keep your terminal free.

• Wait about 30 seconds for the init container to finish, then check that all services are healthy by running:

docker compose ps

You should see postgres, airflow-webserver, and airflow-scheduler with a status of running (or healthy for the webserver). The airflow-init container shows Exited (0) which is expected.

Containers not starting?

If the webserver shows unhealthy, give it another 30 seconds. The health check has a 30-second start period before it begins checking.

If airflow-init exited with a non-zero code, check logs with docker compose logs airflow-init. The most common cause is Postgres not being ready yet.

If you see port conflicts on 8080, another service is using that port. Stop it or change the port mapping in docker-compose.yml.

help me debug my Airflow Docker Compose containers not starting

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

You should see the Airflow DAGs list page with ecommerce_batch_pipeline listed. It will be paused (toggle is off) by default.

Why is the DAG paused?

Your docker-compose.yml sets AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'. This prevents DAGs from automatically running the moment they are detected. You control when to trigger them.

Trigger the DAG and verify end-to-end

Time to run the full pipeline. You will trigger a single DAG run and watch all four tasks execute in sequence.

• In the Airflow UI, click the toggle switch next to ecommerce_batch_pipeline to unpause the DAG.
• Click the play icon (triangle) in the Actions column on the right side of the DAG row.
• Select Trigger DAG from the dropdown.

Before you watch the result, think about what you expect to happen. The DAG has catchup=True and max_active_runs=1. Do you think it will run one date or try to backfill all 30 days at once?

• Click on the DAG name ecommerce_batch_pipeline to open the DAG detail view.
• Switch to the Graph tab to watch task execution.

Airflow begins backfilling from the start_date (June 10, 2026), processing one date per run. With max_active_runs=1, it queues remaining dates and processes them sequentially.

Monitor the first run only

A full backfill of 30 days takes time (each day generates and transforms 170K+ records). For verification, wait for the first run's tasks to turn green (dark green outlines in the Graph view). You can pause the DAG again after confirming the first run succeeds to stop further backfill.

• Wait for all four tasks in the first run to show a dark green border (success state). This typically takes 3-5 minutes depending on your machine and network speed.

All tasks should turn green. If a task turns red (failed), click on it, then click Log to see the error details.

Task turning red?

If generate_data fails, the most common cause is incorrect AWS credentials in your .env file. Check that AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are correct.

If spark_transform fails, check that Java is installed correctly in the container. Run docker compose exec airflow-scheduler java -version to confirm.

If load_snowflake fails, verify your Snowflake credentials in .env and that the storage integration is correctly configured.

my Airflow DAG task is failing and I need help reading the logs

• Once the first run completes successfully, switch to the Snowflake web console and run this query to verify data loaded:

-- Check that fact_orders has rows from the pipeline
SELECT COUNT(*) AS total_orders FROM analytics.fact_orders;

You should see a row count in the thousands (one day of generated orders after deduplication).

• Run this query to confirm the SCD Type 2 customer dimension is populated:

-- Verify SCD Type 2 customer records
SELECT * FROM analytics.dim_customers
WHERE is_current = TRUE
LIMIT 10;

You should see customer rows with is_current = TRUE, a valid_from date, and valid_to set to 9999-12-31. This proves the SCD Type 2 pattern is working.

Why does this matter?

You now have a fully orchestrated pipeline. Airflow handles the scheduling, retry logic, and dependency ordering. If the Spark job fails due to a network timeout, Airflow retries it up to 3 times with exponential backoff.

If the Snowflake load partially completes, the idempotent DELETE-INSERT pattern ensures a clean re-run. This is the architecture a Senior Data Engineer builds and defends in interviews.

Your pipeline runs end-to-end with a single docker compose up command. Data flows from generation through transformation into a Snowflake star schema, orchestrated by Airflow with built-in retries and idempotency. Next up, you will push this further with data quality gates in the Secret Mission.

Secret mission

Add Data Quality Gates

Your pipeline works end-to-end. But what happens when bad data sneaks through? In production, a senior data engineer never loads unchecked data into the warehouse. Add a validation gate between your Spark transformation and the Snowflake load that halts the pipeline when critical quality thresholds are breached.

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 an S3 bucket with minimal storage costs and a Snowflake trial that expires after 30 days.

Cost warning

Your S3 bucket charges approximately $0.023/GB/month for stored data. For this project (a few GB of Parquet files), expect less than $1/month.

Your Snowflake trial uses free credits and expires 30 days after sign-up. If you are not continuing to the next project, choose Pause or Delete to avoid unexpected S3 charges.

Resources you used:

• S3 bucket (ecommerce-pipeline-{initials}-dev) containing raw and curated Parquet data.
• Snowflake database (ECOMMERCE_DW) and warehouse (PIPELINE_WH).
• Docker containers and volumes (Airflow webserver, scheduler, Postgres metadata database).
• Local project files (virtual environment, source code, DAGs, Docker configuration, logs).

Keep everything running

No action needed. Choose this if you are still actively building on this pipeline or plan to start the next project in the series soon.

• Docker containers continue running Airflow on port 8080.
• S3 data persists at approximately $0.023/GB/month.
• Snowflake trial countdown continues regardless of activity.

Pause - I'll come back to this later

Shut down running processes to free up memory and Docker resources. Your project files and S3 data remain intact for when you return.

• Stop the Airflow containers by running this command from your project folder:
• S3 data remains in place at minimal cost (under $1/month).
• Your Snowflake trial continues its 30-day countdown whether you use it or not.

docker compose down

Resuming later

When you return, run docker compose up -d from your project folder to restart Airflow. Your DAGs, source code, and Docker volumes are preserved.

Delete - I don't want to use this again

Remove all project resources completely. This is permanent and cannot be undone.

• Stop and remove all Docker containers, networks, and volumes by running:

docker compose down -v

• Delete the S3 bucket and all its contents by running:

aws s3 rb s3://[[S3BUCKET="ecommerce-pipeline-your-initials-dev"]] --force

• Drop the Snowflake database and warehouse. Open a worksheet in the Snowflake web console and run:

DROP DATABASE IF EXISTS ECOMMERCE_DW;
DROP WAREHOUSE IF EXISTS PIPELINE_WH;

• Deactivate and remove the Python virtual environment by running:

deactivate
rm -rf .venv

• Delete the local project folder. On macOS/Linux:

cd ~/Desktop
rm -rf ecommerce-pipeline

On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete.

What about the Snowflake trial account?

Snowflake trial accounts cannot be self-deleted through the web UI. The trial expires automatically after 30 days or when the $400 in free credits is used up, whichever comes first.

Dropping the database and warehouse above ensures no queries run against your credits.

That's a Wrap!

That's a Wrap!

Nice work! You've just built a production-grade batch data pipeline that processes millions of e-commerce order events from raw ingestion through to a queryable analytics warehouse.

You've learned how to:

• Built a multi-layer S3 data lake and PySpark transformation pipeline that processes 5M+ e-commerce order events with deduplication via window functions, late-arriving data re-partitioning, and lineage tracking.
• Designed and loaded a Snowflake star schema with SCD Type 2 customer dimension, idempotent MERGE/DELETE-INSERT patterns, and proper surrogate key management.
• Orchestrated the full pipeline with Dockerized Airflow including retries, incremental processing, catchup backfill, and depends_on_past sequential execution.
• Secret Mission: Added data quality gates that validate curated data between the Spark transform and Snowflake load, preventing bad records from reaching the warehouse.

Ready to quiz yourself?