Design a Batch ETL Pipeline to Process E-commerce Transactions

E-commerce platforms generate massive volumes of transactional data, including orders, customer profiles, and inventory updates. A Batch ETL (Extract, Transform, Load) pipeline is typically used to process this data on a scheduled basis (e.g., nightly) to feed analytical dashboards and generate business reports.

Continuing with our modern data stack, this design leverages AWS, PySpark, and Snowflake to build a robust batch processing architecture.

1. High-Level Architecture Diagram

flowchart LR
    subgraph "Data Sources (OLTP)"
        A1[("Amazon RDS<br>MySQL - Orders")]
        A2[("Amazon RDS<br>PostgreSQL - Users")]
    end

    subgraph "Ingestion (Extract)"
        B1["AWS DMS<br>(Database Migration Service)"]
    end

    subgraph "Data Lake Storage (AWS S3)"
        C1[("S3 - Raw/Bronze")]
        C2[("S3 - Processed/Silver")]
    end

    subgraph "Processing (Transform)"
        D1["Amazon EMR / AWS Glue<br>(PySpark Batch Job)"]
    end

    subgraph "Data Warehouse (Load/Gold)"
        E1[("Snowflake Staging")]
        E2[("Snowflake Star Schema")]
    end

    subgraph "Consumption"
        F1["BI Dashboards"]
    end

    A1 -->|Full Load / CDC| B1
    A2 -->|Full Load / CDC| B1

    B1 -->|Write CSV/JSON| C1

    C1 -->|Read Batch| D1
    D1 -->|Clean, Join, Agg| D1
    D1 -->|Write Parquet| C2

    C2 -->|COPY INTO via Airflow or Task| E1
    E1 -->|dbt / SQL| E2

    E2 --> F1

2. Component Breakdown

Step 1: Extract (Data Sources to Data Lake)

E-commerce data usually resides in Relational Databases (OLTP systems) designed for fast transactions, not heavy analytics.

• Tool: AWS Database Migration Service (DMS) or scheduled AWS Glue Crawlers.
• Function: Extracts data from the source databases (e.g., Orders, Customers, Products tables) and dumps it into the Amazon S3 Raw Bucket (Bronze Zone). This can be done as a nightly full load (for smaller tables like Products) or an incremental load (CDC - Change Data Capture) for massive tables like Orders.

Step 2: Transform (PySpark Batch Processing)

Once the raw data lands in S3, a scheduled PySpark job (often orchestrated by Apache Airflow or AWS Step Functions) wakes up to process it.

• Tool: PySpark running on Amazon EMR or AWS Glue.
• Operations Performed:
  • Data Cleansing: Handling missing values, standardizing date formats, and dropping duplicate records.
  • Joining Data: Joining the Orders data with Customers and Products to create a denormalized, wide dataset.
  • Business Logic: Calculating metrics like total_order_value (quantity * price), applying discounts, and standardizing currency.
  • Load to Silver Zone: The transformed DataFrames are written to the S3 Processed Bucket (Silver Zone) in highly compressed Parquet format, partitioned by date (e.g., year=2026/month=05/day=21).

Step 3: Load (Data Warehousing in Snowflake)

The cleaned Parquet files are now ready to be loaded into the Data Warehouse for modeling and querying.

• Tool: Snowflake.
• Workflow:
  1. A scheduler (like Airflow or a Snowflake Task) runs a COPY INTO command to ingest the Parquet files from S3 into a Staging Table in Snowflake.
  2. Once in Snowflake, SQL scripts (often managed by tools like dbt) transform the flat staging data into a Star Schema (Fact and Dimension tables), which is the optimal structure for BI tools.
     • Fact Table: fact_orders
     • Dimension Tables: dim_customers, dim_products, dim_time.

Step 4: Consumption

Data Analysts query the Star Schema in Snowflake using BI tools (Tableau, Power BI) to generate reports on daily revenue, customer lifetime value (LTV), and best-selling products.

3. Code Snippets

PySpark: Batch Transformation

This PySpark script reads raw data from S3, performs an inner join between orders and customers, calculates the total amount, and writes the output as partitioned Parquet files.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, round, current_date

# Initialize Spark Session for Batch Processing
spark = SparkSession.builder \
    .appName("EcommerceBatchETL") \
    .getOrCreate()

# 1. Read Raw Data from S3
orders_df = spark.read.csv("s3://ecommerce-datalake/raw/orders/date=2026-05-21/", header=True, inferSchema=True)
customers_df = spark.read.csv("s3://ecommerce-datalake/raw/customers/", header=True, inferSchema=True)
products_df = spark.read.csv("s3://ecommerce-datalake/raw/products/", header=True, inferSchema=True)

# 2. Transform: Join Orders with Customers and Products
enriched_orders = orders_df \
    .join(customers_df, orders_df.customer_id == customers_df.id, "inner") \
    .join(products_df, orders_df.product_id == products_df.id, "inner") \
    .select(
        orders_df.order_id,
        customers_df.customer_name,
        customers_df.region,
        products_df.product_name,
        orders_df.quantity,
        products_df.price,
        orders_df.order_date
    )

# 3. Transform: Calculate Total Order Value
final_df = enriched_orders.withColumn(
    "total_amount", 
    round(col("quantity") * col("price"), 2)
)

# 4. Write Processed Data to S3 in Parquet (Partitioned by Date)
final_df.write \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://ecommerce-datalake/processed/enriched_orders/")

spark.stop()

Snowflake: Loading and Modeling

This SQL script demonstrates how to load the processed Parquet data into a staging table and then insert it into a Fact table.

-- 1. Setup External Stage (Assuming Storage Integration is already created)
CREATE OR REPLACE STAGE ecommerce_s3_stage
  STORAGE_INTEGRATION = s3_int
  URL = 's3://ecommerce-datalake/processed/enriched_orders/'
  FILE_FORMAT = (TYPE = PARQUET);

-- 2. Create Staging Table
CREATE OR REPLACE TRANSIENT TABLE stg_enriched_orders (
    order_id STRING,
    customer_name STRING,
    region STRING,
    product_name STRING,
    quantity NUMBER,
    price FLOAT,
    total_amount FLOAT,
    order_date DATE
);

-- 3. Load Data from S3 into Staging (Batch Load)
COPY INTO stg_enriched_orders
FROM @ecommerce_s3_stage
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

-- 4. Create Fact Table (Star Schema)
CREATE TABLE IF NOT EXISTS fact_orders (
    order_id STRING PRIMARY KEY,
    customer_name STRING,
    product_name STRING,
    total_amount FLOAT,
    order_date DATE
);

-- 5. Insert New Data into Fact Table (UPSERT / MERGE Pattern)
MERGE INTO fact_orders f
USING stg_enriched_orders s
ON f.order_id = s.order_id
WHEN MATCHED THEN 
    UPDATE SET 
        f.total_amount = s.total_amount,
        f.order_date = s.order_date
WHEN NOT MATCHED THEN 
    INSERT (order_id, customer_name, product_name, total_amount, order_date)
    VALUES (s.order_id, s.customer_name, s.product_name, s.total_amount, s.order_date);

-- 6. Clean up Staging Table
TRUNCATE TABLE stg_enriched_orders;