Build an ETL Pipeline with PySpark

Process messy CSV data into clean, analytics-ready Parquet files.

Introduction

30 Second Summary

Every online order you place generates data. Behind the scenes, companies run automated pipelines to clean, organize, and analyze millions of these messy records so they can answer questions like "what sold best last month?"

In this project, you will build an ETL pipeline with PySpark that takes messy e-commerce order data and transforms it into clean, analytics-ready files.

What You'll Build

A complete ETL (Extract-Transform-Load) pipeline that processes raw CSV files with inconsistent dates, missing values, and dirty formatting into structured Parquet files ready for analytics.

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

  • A PySpark script that reads and processes raw CSV data through extract, transform, and load stages.
  • Transformation logic that cleans dates, handles nulls, standardizes text, and derives new columns.
  • Spark SQL queries that generate business insights like monthly revenue and top products.
  • Secret Mission: Add a data quality report that flags anomalies and logs statistics about the pipeline run.

Are there any prerequisites?

You should have basic Python knowledge (variables, functions, loops). No prior experience with PySpark or data engineering is needed.

Before We Start

Before We Start...

Install Java and Python

PySpark is built on top of Apache Spark, which runs on the Java Virtual Machine (JVM). Before you can process any data with PySpark, your machine needs both Java and Python installed and configured.

In this step, you will install Java 17, verify Python is ready, and confirm your environment variables are set. This ensures PySpark can find the tools it depends on when you run your pipeline later.

In this step, get ready to:
  • Install Java 17 and verify it with java -version.
  • Verify Python 3.10+ is installed and accessible.
  • Confirm the JAVA_HOME environment variable is set correctly.
Install Java 17

PySpark 4.1.2 requires Java 17 or later. You will install the Microsoft Build of OpenJDK using the Windows Package Manager.

  • Press the Windows key to open your search bar.
  • Type Command Prompt and press Enter to open it.

Why Command Prompt?

You can use either Command Prompt or PowerShell for this project. Both work with winget and PySpark. This project shows Command Prompt examples, but PowerShell equivalents are noted where they differ.

  • Check whether Java is already installed by running this command:
java -version

✔️ I see version 17 or higher

Your output shows something like openjdk version "17.0.x". Java 17 is ready. Skip ahead to the next substep.

ⓧ I see an older version

You have Java installed, but PySpark 4.1.2 requires version 17 or later. Install Java 17 alongside your existing version by running this command:

winget install Microsoft.OpenJDK.17

What does this command do?

This uses the Windows Package Manager (winget) to download and install the Microsoft Build of OpenJDK 17. The installer adds Java to your system PATH automatically.

The installer may prompt you to accept license terms. Type Y and press Enter if prompted.

  • Close your Command Prompt window and open a new one (this refreshes your PATH).
  • Verify the installation by running:
java -version

You should see output containing openjdk version "17.0.x".

ⓧ Command not found

Java is not installed on your machine. Install Java 17 by running this command:

winget install Microsoft.OpenJDK.17

What does this command do?

This uses the Windows Package Manager (winget) to download and install the Microsoft Build of OpenJDK 17. The installer adds Java to your system PATH automatically.

The installer may prompt you to accept license terms. Type Y and press Enter if prompted.

  • Close your Command Prompt window and open a new one (this refreshes your PATH).
  • Verify the installation by running:
java -version

You should see output containing openjdk version "17.0.x".

Still seeing command not found?

Make sure you opened a fresh Command Prompt after the install finished. If winget itself is not recognized, your Windows version may be too old. You need Windows 10 or later with the App Installer package from the Microsoft Store.

Help me troubleshoot my Java installation on Windows.

Verify Python 3.10+ is installed

PySpark 4.1.2 requires Python 3.10 or above. Let's check whether your machine already has a compatible version.

  • Check your Python version by running this command:
python --version

✔️ I see version 3.10 or higher

Your output shows something like Python 3.10.x or higher. Python is ready. Move on to the next substep.

ⓧ I see a version lower than 3.10

Your Python version is too old for PySpark 4.1.2. You need to install Python 3.10 or later.

  • Open your browser and go to python.org/downloads.
  • Click the Download Python button to get the latest version.
  • Run the installer. On the first screen, check the box labeled Add python.exe to PATH before clicking Install Now.
  • Close your Command Prompt and open a new one.
  • Run python --version again to confirm the new version appears.

Add to PATH is critical

If you miss the Add python.exe to PATH checkbox, your terminal will not find the python command. You would need to uninstall and reinstall Python with that option checked.

ⓧ Command not found

Python is not installed on your machine. You need to download it from the official website.

  • Open your browser and go to python.org/downloads.
  • Click the Download Python button to get the latest version.
  • Run the installer. On the first screen, check the box labeled Add python.exe to PATH before clicking Install Now.
  • Close your Command Prompt and open a new one.
  • Verify by running:
python --version

You should see Python 3.10.x or higher.

Still seeing command not found after installing?

Make sure you checked Add python.exe to PATH during installation. If you missed it, uninstall Python from Settings > Apps and reinstall with that option checked.

Help me get Python working on Windows.

Verify JAVA_HOME is set correctly

PySpark needs the JAVA_HOME environment variable to locate your Java installation. Many tools like Maven and Gradle also depend on it. Let's confirm it is set.

  • Check your JAVA_HOME variable by running this command in Command Prompt:
echo %JAVA_HOME%

Using PowerShell instead?

If you are using PowerShell, the equivalent command is:

echo $env:JAVA_HOME

✔️ I see a file path

Your output shows a path like C:\Program Files\Microsoft\jdk-17.x.x. This means JAVA_HOME is configured and PySpark will be able to find Java.

ⓧ It shows %JAVA_HOME% or is empty

The JAVA_HOME variable is not set. You need to add it manually.

  • Press the Windows key and type Environment Variables.
  • Click Edit the system environment variables.
  • Click Environment Variables at the bottom of the dialog.
  • Under System variables, click New.
  • Set Variable name to JAVA_HOME.
  • Set Variable value to the path where Java 17 was installed. The default path is C:\Program Files\Microsoft\jdk-17.0.x.x-hotspot (check your C:\Program Files\Microsoft\ folder for the exact folder name).
  • Click OK on all open dialogs to save.
  • Close your Command Prompt and open a new one.
  • Verify by running:
echo %JAVA_HOME%

You should now see the path to your Java installation.

JAVA_HOME still empty after setting it?

You must open a brand new Command Prompt window after changing environment variables. Existing windows will not pick up the change.

Help me set JAVA_HOME correctly on Windows.

Your machine now has Java 17 and Python ready. Next up, you will create the project folder structure and install PySpark itself.

Set Up the Project and Install PySpark

With Java and Python ready on your machine, you have the foundation for running PySpark. Now you need a workspace that keeps your raw data, processing scripts, and cleaned outputs organized.

In this step, you'll create the project folder structure, install PySpark into an isolated virtual environment, and prepare the sample data your pipeline will process.

In this step, get ready to:
  • Create the ecommerce_etl/ project folder with organized subdirectories.
  • Install PySpark 4.1.2 in a Python virtual environment.
  • Create a sample CSV file with intentionally messy data.
Create the project folder structure

A well-organized folder structure separates concerns: raw input data, processing scripts, and cleaned output each get their own home.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create the project folder and its subdirectories by running these commands:

macOS

mkdir -p ecommerce_etl/data/raw ecommerce_etl/data/processed ecommerce_etl/src

Windows

mkdir ecommerce_etl\data\raw, ecommerce_etl\data\processed, ecommerce_etl\src

What does this create?

  • data/raw/ holds the original messy input files.
  • data/processed/ is where your pipeline will write cleaned output.
  • src/ contains your Python pipeline scripts.
  • Move into the project folder by running:
cd ecommerce_etl
  • Confirm the folders exist by running:
ls

You should see data and src listed in the output.

Don't see the folders?

Make sure you ran the mkdir command from your Desktop, not from inside another folder. Run pwd (macOS) or cd (Windows) to check your current location.

help me figure out why my project folders weren't created

Create a virtual environment and install PySpark

A virtual environment isolates your project's packages from the rest of your system. This prevents version conflicts between projects.

  • Create a virtual environment by running:
python -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms you're working inside the virtual environment.

Seeing a permissions error on Windows?

PowerShell may block script execution by default. Run this command to allow it, then try activating again:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

help me fix the virtual environment activation error

  • Install PySpark by running:
pip install pyspark==4.1.2

This downloads PySpark and its dependencies. The install may take a minute or two.

  • Verify the installation succeeded by running:
python -c "import pyspark; print(pyspark.__version__)"

You should see 4.1.2 printed in the terminal.

Seeing an import error?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, run the activation command again.

Also confirm JAVA_HOME is still set correctly. PySpark requires Java to run.

help me fix PySpark not importing correctly

Create the sample data file

Your pipeline needs raw data to process. You'll create a CSV file with intentionally messy e-commerce order data that mimics real-world data problems.

  • Open VS Code in the project folder by running:
code .

Command not found?

If code isn't recognized, open VS Code manually. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar. Type Visual Studio Code and press Enter. Then open the ecommerce_etl folder from File > Open Folder.

  • In the VS Code Explorer sidebar, expand the data folder, then raw.
  • Right-click on raw and select New File.
  • Name the file orders.csv and press Enter.
  • Paste the following data into orders.csv:
order_id,customer_id,product,amount,status,order_date,country
1001,C001,Laptop,999.99,Completed,2024-01-15,United States
1002,C002,Mouse,29.99,completed,2024-01-16,united states
1003,C003,Keyboard,79.99,COMPLETED,01/20/2024,Canada
1004,,Monitor,349.99,Pending,2024-01-22,United Kingdom
1005,C005,Headphones,-49.99,Completed,2024-02-01,United States
1006,C006,Webcam,89.99,Shipped,2024-02-03,Canada
1007,C007,Laptop,1299.99,Completed,2024-02-10,United States
1008,C008,Mouse,24.99,cancelled,2024-02-14,Germany
1009,C009,Tablet,499.99,Completed,2024-03-01,United States
1010,C010,Chair,259.99,Completed,03/05/2024,Canada
1011,C011,Desk,549.99,Pending,2024-03-10,United Kingdom
1012,C012,Monitor,399.99,Completed,2024-03-15,United States
,C013,Keyboard,69.99,Shipped,2024-03-20,Germany
1014,C014,Laptop,1499.99,Completed,2024-04-01,United States
1015,C015,Headphones,199.99,Completed,2024-04-05,Canada
1016,C016,Webcam,59.99,Returned,2024-04-10,United States
1017,C017,Tablet,599.99,Completed,2024-04-15,United Kingdom
1018,C018,Chair,299.99,Shipped,2024-05-01,Germany
1019,C019,Desk,449.99,Completed,2024-05-10,United States
1020,C020,Monitor,279.99,Completed,2024-05-15,Canada

What's wrong with this data?

This CSV has the same problems you'd find in real production data. Your pipeline will need to handle all of them:

  • Mixed date formats: 2024-01-15 vs 01/20/2024.
  • Inconsistent casing: Completed, completed, COMPLETED.
  • Missing values: empty customer_id (row 4) and empty order_id (row 13).
  • Invalid data: a negative amount (-49.99 in row 5).
  • Save orders.csv by pressing Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

Here's the complete orders.csv file. Make sure yours matches exactly:

order_id,customer_id,product,amount,status,order_date,country
1001,C001,Laptop,999.99,Completed,2024-01-15,United States
1002,C002,Mouse,29.99,completed,2024-01-16,united states
1003,C003,Keyboard,79.99,COMPLETED,01/20/2024,Canada
1004,,Monitor,349.99,Pending,2024-01-22,United Kingdom
1005,C005,Headphones,-49.99,Completed,2024-02-01,United States
1006,C006,Webcam,89.99,Shipped,2024-02-03,Canada
1007,C007,Laptop,1299.99,Completed,2024-02-10,United States
1008,C008,Mouse,24.99,cancelled,2024-02-14,Germany
1009,C009,Tablet,499.99,Completed,2024-03-01,United States
1010,C010,Chair,259.99,Completed,03/05/2024,Canada
1011,C011,Desk,549.99,Pending,2024-03-10,United Kingdom
1012,C012,Monitor,399.99,Completed,2024-03-15,United States
,C013,Keyboard,69.99,Shipped,2024-03-20,Germany
1014,C014,Laptop,1499.99,Completed,2024-04-01,United States
1015,C015,Headphones,199.99,Completed,2024-04-05,Canada
1016,C016,Webcam,59.99,Returned,2024-04-10,United States
1017,C017,Tablet,599.99,Completed,2024-04-15,United Kingdom
1018,C018,Chair,299.99,Shipped,2024-05-01,Germany
1019,C019,Desk,449.99,Completed,2024-05-10,United States
1020,C020,Monitor,279.99,Completed,2024-05-15,Canada

Your workspace is ready with organized folders, PySpark installed, and raw data waiting to be processed. Next up, you'll write your first SparkSession and load this messy CSV into a DataFrame.

Create Your SparkSession and Load Data

Your project structure is ready and PySpark is installed. Now it's time to write your first PySpark code.

Every PySpark program starts with a SparkSession. In this step, you'll create one and use it to load your raw CSV data into a DataFrame for inspection.

In this step, get ready to:
  • Create src/etl_pipeline.py with a SparkSession builder function.
  • Add an extract function that reads CSV data into a DataFrame.
  • Run the pipeline and observe the raw data output.
Create the pipeline file and initialize a SparkSession

A SparkSession is the single entry point to all PySpark operations. It manages your connection to Spark and provides methods for reading data, creating DataFrames, and running SQL queries.

  • In VS Code, right-click the src folder in the Explorer sidebar.
  • Select New File and name it etl_pipeline.py.
  • Add the following imports and SparkSession builder function:
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, to_date, year, month, when, coalesce, lit, sum as spark_sum, count, avg
import os


def create_spark_session():
    """Create and return a SparkSession."""
    return SparkSession.builder \
        .appName("Ecommerce_ETL_Pipeline") \
        .config("spark.sql.adaptive.enabled", "true") \
        .getOrCreate()

What does this code do?

The imports bring in SparkSession for connecting to Spark, plus transformation functions like col, lower, and to_date that you'll use in later steps. The os module handles cross-platform file paths.

The create_spark_session() function uses the builder pattern to configure and start Spark. .appName() names your application for identification. .config("spark.sql.adaptive.enabled", "true") enables adaptive query execution for better performance. .getOrCreate() either creates a new session or reuses an existing one.

  • Save src/etl_pipeline.py.

You won't see a change yet. The next chunk adds the function that actually reads data.

Add the extract function to read CSV data

The extract phase is the first stage of any ETL pipeline. It loads raw data from a source into memory so you can work with it.

  • In src/etl_pipeline.py, add the extract() function below create_spark_session():
def extract(spark, input_path):
    """Extract: Read raw CSV data into a DataFrame."""
    print("=" * 50)
    print("EXTRACT PHASE")
    print("=" * 50)

    df = spark.read.csv(input_path, header=True, inferSchema=True)

    print(f"\nLoaded {df.count()} rows from {input_path}")
    print("\nSchema:")
    df.printSchema()
    print("\nSample data:")
    df.show(5, truncate=False)

    return df

What does this code do?

This function reads the raw CSV file into a DataFrame. header=True tells Spark the first row contains column names. inferSchema=True makes Spark detect data types automatically instead of treating everything as strings.

After loading, the function prints three things to help you understand the data. .count() returns the total number of rows. .printSchema() displays each column's name and inferred data type. .show(5, truncate=False) displays the first 5 rows without cutting off long values.

  • Save src/etl_pipeline.py.

You won't see a change yet. The next chunk adds the main function that ties everything together and lets you run the script.

Add the main function and run the pipeline

The main() function orchestrates the pipeline. It creates the SparkSession, calls extract, and ensures Spark shuts down properly with a try/finally block.

  • Add the main() function at the bottom of src/etl_pipeline.py:
def main():
    """Main ETL pipeline orchestration."""
    print("\n" + "#" * 50)
    print("  ECOMMERCE ETL PIPELINE - STARTING")
    print("#" * 50)

    # Paths
    input_path = os.path.join("data", "raw", "orders.csv")
    output_orders_path = os.path.join("data", "processed", "orders")
    output_summary_path = os.path.join("data", "processed", "summary")

    # Create SparkSession
    spark = create_spark_session()

    try:
        # Extract
        raw_df = extract(spark, input_path)

    except Exception as e:
        print(f"\nPIPELINE FAILED: {e}")
        raise

    finally:
        spark.stop()
        print("\nSpark session closed.")


if __name__ == "__main__":
    main()

What does this code do?

The main() function defines three file paths using os.path.join() for cross-platform compatibility. It then creates a SparkSession and calls your extract() function.

The try/finally block ensures spark.stop() always runs, even if something goes wrong. This releases memory and shuts down Spark cleanly. The if __name__ == "__main__" guard means the pipeline only runs when you execute the file directly.

  • Save src/etl_pipeline.py.
  • Run the pipeline by executing this command in your terminal:
python src/etl_pipeline.py

You should see output showing the pipeline banner, then the schema (column names and types) and the first 5 rows of raw data.

What should I see?

The terminal displays a schema listing columns like order_id (integer), customer_id (string), amount (double), and order_date (string). The sample rows show the messy values you'll clean in the next step: inconsistent casing in status (Completed, completed, COMPLETED), mixed date formats, and a negative amount.

Seeing an error when running the script?

  • If you see a Java-related error, check that JAVA_HOME is set correctly. Run echo %JAVA_HOME% (Command Prompt) or echo $env:JAVA_HOME (PowerShell) and confirm it points to your Java 17 installation.
  • If you see a path-related error mentioning the CSV file, make sure you're running the command from the ecommerce_etl folder (the one containing the data and src folders).
  • If you see an import error for pyspark, confirm your virtual environment is activated. You should see (venv) at the start of your terminal prompt.

Help me troubleshoot my PySpark pipeline error

✔️ Awesome, I've got everything!

Great. Double check you've saved src/etl_pipeline.py and you're ready for the next step.

ⓧ I'd like to double check the full code

Here's the complete src/etl_pipeline.py at this point:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, to_date, year, month, when, coalesce, lit, sum as spark_sum, count, avg
import os


def create_spark_session():
    """Create and return a SparkSession."""
    return SparkSession.builder \
        .appName("Ecommerce_ETL_Pipeline") \
        .config("spark.sql.adaptive.enabled", "true") \
        .getOrCreate()


def extract(spark, input_path):
    """Extract: Read raw CSV data into a DataFrame."""
    print("=" * 50)
    print("EXTRACT PHASE")
    print("=" * 50)

    df = spark.read.csv(input_path, header=True, inferSchema=True)

    print(f"\nLoaded {df.count()} rows from {input_path}")
    print("\nSchema:")
    df.printSchema()
    print("\nSample data:")
    df.show(5, truncate=False)

    return df


def main():
    """Main ETL pipeline orchestration."""
    print("\n" + "#" * 50)
    print("  ECOMMERCE ETL PIPELINE - STARTING")
    print("#" * 50)

    # Paths
    input_path = os.path.join("data", "raw", "orders.csv")
    output_orders_path = os.path.join("data", "processed", "orders")
    output_summary_path = os.path.join("data", "processed", "summary")

    # Create SparkSession
    spark = create_spark_session()

    try:
        # Extract
        raw_df = extract(spark, input_path)

    except Exception as e:
        print(f"\nPIPELINE FAILED: {e}")
        raise

    finally:
        spark.stop()
        print("\nSpark session closed.")


if __name__ == "__main__":
    main()

Your pipeline is loading data successfully. Next up, you'll transform this messy data by cleaning nulls, standardizing text, and parsing dates into a consistent format.

Transform and Clean the Data

Your extract phase pulled the raw data into a DataFrame. But if you look at the output from the last step, you'll notice problems. Some rows have missing IDs, one order has a negative amount, and the status column has "Completed", "completed", and "COMPLETED" as three separate values.

This step tackles the "T" in ETL. You'll build a transform function that cleans the messy data and enriches it with new columns for easier analysis downstream.

In this step, get ready to:
  • Drop rows with null IDs and negative amounts.
  • Standardize text and parse dates into a consistent format.
  • Derive new columns like order year, month, and amount category.
Drop rows with null IDs and negative amounts

Real-world data is never perfect. Before you can analyze orders, you need to remove records that are fundamentally broken. A row missing its order_id can't be tracked. A negative amount doesn't represent a real purchase.

  • In src/etl_pipeline.py, find the main() function. Add the following transform() function directly above it:
def transform(df):
    """Transform: Clean and enrich the data."""
    print("\n" + "=" * 50)
    print("TRANSFORM PHASE")
    print("=" * 50)

    initial_count = df.count()

    # Drop rows with null order_id or customer_id
    df_clean = df.dropna(subset=["order_id", "customer_id"])

    # Filter out negative amounts
    df_clean = df_clean.filter(col("amount") > 0)

    cleaned_count = df_clean.count()
    print(f"\nRows removed during cleaning: {initial_count - cleaned_count}")

What does this code do?

  • df.count() captures the total row count before any cleaning happens. This lets you report how many rows were removed.
  • .dropna(subset=["order_id", "customer_id"]) removes any row where either of those columns is null. In your data, row 1004 has no customer_id and one row has no order_id.
  • .filter(col("amount") > 0) keeps only rows with positive amounts. This catches the headphones order with -49.99.

You won't see a change yet. The next substep continues building this function.

Standardize text and parse dates

Your status column has "Completed", "completed", and "COMPLETED" as separate values. The date column mixes two formats. Without standardizing these, any downstream analysis or grouping would treat them as different categories.

  • In src/etl_pipeline.py, find the line that says print(f"\nRows removed during cleaning: {initial_count - cleaned_count}") inside your transform() function. Add the following code below it:
    # Standardize status to lowercase
    df_clean = df_clean.withColumn("status", lower(col("status")))

    # Parse dates - handle multiple formats using coalesce
    df_clean = df_clean.withColumn(
        "order_date_parsed",
        coalesce(
            to_date(col("order_date"), "yyyy-MM-dd"),
            to_date(col("order_date"), "MM/dd/yyyy")
        )
    )

    # Derive year and month columns
    df_clean = df_clean.withColumn("order_year", year(col("order_date_parsed")))
    df_clean = df_clean.withColumn("order_month", month(col("order_date_parsed")))

What does this code do?

  • lower(col("status")) converts every status value to lowercase. Now "Completed", "COMPLETED", and "completed" all become the same value.
  • coalesce() tries each date format in order and returns the first one that parses successfully. Dates like 2024-01-15 match the first format. Dates like 01/20/2024 match the second.
  • year() and month() extract numeric values from the parsed date. These new columns make it easy to partition or group data by time period later.

Why coalesce instead of a single date format?

Production data often arrives with inconsistent formatting. One source might use yyyy-MM-dd while another uses MM/dd/yyyy. The coalesce() pattern handles this gracefully by trying each format in sequence.

You won't see a change yet. The next substep finishes the function.

Add an amount category column

Categorizing continuous values into buckets is a common transformation. Instead of working with raw dollar amounts, you can label each order as "small", "medium", or "large" for quick summaries.

  • Still in the transform() function, find the line df_clean = df_clean.withColumn("order_month", month(col("order_date_parsed"))). Add the following code below it:
    # Add amount category
    df_clean = df_clean.withColumn(
        "amount_category",
        when(col("amount") < 100, "small")
        .when(col("amount") < 500, "medium")
        .otherwise("large")
    )

    # Drop the original order_date and rename parsed date
    df_clean = df_clean.drop("order_date") \
        .withColumnRenamed("order_date_parsed", "order_date")

    print(f"\nRows after cleaning: {cleaned_count}")
    print("\nCleaned data sample:")
    df_clean.show(5, truncate=False)

    return df_clean

What does this code do?

  • when() and .otherwise() work like an if/elif/else chain. Orders under $100 are "small", under $500 are "medium", and everything else is "large".
  • .drop("order_date") removes the original messy date column. Then .withColumnRenamed() renames order_date_parsed to order_date so downstream code can reference the clean date with the original column name.
  • return df_clean passes the fully cleaned DataFrame back to the caller.

Now you need to call your new function from main().

  • In the main() function, find this line:
        # Extract
        raw_df = extract(spark, input_path)
  • Add the following lines directly below raw_df = extract(spark, input_path):
        # Transform
        clean_df = transform(raw_df)
  • Save src/etl_pipeline.py.
  • Run the script by executing this command in your terminal:
python src/etl_pipeline.py
  • You should see the TRANSFORM PHASE header, a message saying Rows removed during cleaning: 3, and a sample of the cleaned data with new columns like order_year, order_month, and amount_category.

Seeing an error when you run the script?

  • If you see an indentation error, make sure all lines inside transform() are indented by exactly 4 spaces from the function definition.
  • If you see a NameError for col or lower, check that your imports at the top of the file include from pyspark.sql.functions import col, lower, to_date, year, month, when, coalesce.
  • If the script runs but shows 0 rows removed, make sure you placed the transform() function above main() and that main() calls transform(raw_df).

Why is my transform function throwing an error?

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the full src/etl_pipeline.py at this point:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, to_date, year, month, when, coalesce, lit, sum as spark_sum, count, avg
import os


def create_spark_session():
    """Create and return a SparkSession."""
    return SparkSession.builder \
        .appName("Ecommerce_ETL_Pipeline") \
        .config("spark.sql.adaptive.enabled", "true") \
        .getOrCreate()


def extract(spark, input_path):
    """Extract: Read raw CSV data into a DataFrame."""
    print("=" * 50)
    print("EXTRACT PHASE")
    print("=" * 50)

    df = spark.read.csv(input_path, header=True, inferSchema=True)

    print(f"\nLoaded {df.count()} rows from {input_path}")
    print("\nSchema:")
    df.printSchema()
    print("\nSample data:")
    df.show(5, truncate=False)

    return df


def transform(df):
    """Transform: Clean and enrich the data."""
    print("\n" + "=" * 50)
    print("TRANSFORM PHASE")
    print("=" * 50)

    initial_count = df.count()

    # Drop rows with null order_id or customer_id
    df_clean = df.dropna(subset=["order_id", "customer_id"])

    # Filter out negative amounts
    df_clean = df_clean.filter(col("amount") > 0)

    cleaned_count = df_clean.count()
    print(f"\nRows removed during cleaning: {initial_count - cleaned_count}")

    # Standardize status to lowercase
    df_clean = df_clean.withColumn("status", lower(col("status")))

    # Parse dates - handle multiple formats using coalesce
    df_clean = df_clean.withColumn(
        "order_date_parsed",
        coalesce(
            to_date(col("order_date"), "yyyy-MM-dd"),
            to_date(col("order_date"), "MM/dd/yyyy")
        )
    )

    # Derive year and month columns
    df_clean = df_clean.withColumn("order_year", year(col("order_date_parsed")))
    df_clean = df_clean.withColumn("order_month", month(col("order_date_parsed")))

    # Add amount category
    df_clean = df_clean.withColumn(
        "amount_category",
        when(col("amount") < 100, "small")
        .when(col("amount") < 500, "medium")
        .otherwise("large")
    )

    # Drop the original order_date and rename parsed date
    df_clean = df_clean.drop("order_date") \
        .withColumnRenamed("order_date_parsed", "order_date")

    print(f"\nRows after cleaning: {cleaned_count}")
    print("\nCleaned data sample:")
    df_clean.show(5, truncate=False)

    return df_clean


def main():
    """Main ETL pipeline orchestration."""
    print("\n" + "#" * 50)
    print("  ECOMMERCE ETL PIPELINE - STARTING")
    print("#" * 50)

    # Paths
    input_path = os.path.join("data", "raw", "orders.csv")
    output_orders_path = os.path.join("data", "processed", "orders")
    output_summary_path = os.path.join("data", "processed", "summary")

    # Create SparkSession
    spark = create_spark_session()

    try:
        # Extract
        raw_df = extract(spark, input_path)

        # Transform
        clean_df = transform(raw_df)

    except Exception as e:
        print(f"\nPIPELINE FAILED: {e}")
        raise

    finally:
        spark.stop()
        print("\nSpark session closed.")


if __name__ == "__main__":
    main()

Your pipeline now has clean, structured data with consistent values. Next up, you'll query this data using Spark SQL and write the results to Parquet files ready for analytics tools.

Analyze with Spark SQL and Load Results

Your data is clean, standardized, and enriched with new columns. The final phase of any ETL pipeline is the "Load" step, where processed data lands in its destination format ready for analytics tools to consume.

In this step, you'll write your cleaned data to Parquet files (the standard output format in production Spark pipelines), use Spark SQL to generate business insights, and run the complete pipeline end-to-end.

In this step, get ready to:
  • Write cleaned orders to partitioned Parquet files.
  • Query cleaned data with Spark SQL for monthly revenue insights.
  • Run the complete pipeline and verify the output.
Write cleaned orders to partitioned Parquet files

Parquet is a columnar file format that stores data efficiently by compressing each column separately. Partitioning splits the output into subfolders based on column values, so analytics tools can skip irrelevant data entirely.

  • In src/etl_pipeline.py, find the transform() function's closing return df_clean statement.
  • Below it (outside the transform function), add the following new load() function:
def load(spark, df_clean, output_orders_path, output_summary_path):
    """Load: Write cleaned data and aggregated summary to Parquet."""
    print("\n" + "=" * 50)
    print("LOAD PHASE")
    print("=" * 50)

    # Write partitioned Parquet
    df_clean.write \
        .mode("overwrite") \
        .partitionBy("order_year", "order_month") \
        .parquet(output_orders_path)

    print(f"\nCleaned orders written to: {output_orders_path}")

What does this code do?

This is the beginning of your load() function, which handles the final ETL stage.

  • .mode("overwrite") replaces existing output if you re-run the pipeline.
  • .partitionBy("order_year", "order_month") creates subfolders like order_year=2024/order_month=1/ so queries on specific months only read the relevant folder.
  • .parquet(output_orders_path) writes the data in Parquet format to the data/processed/orders/ directory.

Why Parquet instead of CSV?

CSV stores data row by row, so reading a single column means scanning every row. Parquet stores data column by column, so analytics queries that only need a few columns skip the rest entirely.

Parquet also embeds the schema (column names and types) directly in the file. No more guessing whether a column is a string or a number.

Register a SQL view and query for business insights

Spark SQL lets you query DataFrames using familiar SQL syntax. You first register the DataFrame as a temporary view, then write standard SQL against it.

  • Add the following code directly below the print(f"\nCleaned orders written to: {output_orders_path}") line, still inside the load() function:
    # Create aggregated summary using Spark SQL
    df_clean.createOrReplaceTempView("clean_orders")

    summary_df = spark.sql("""
        SELECT
            order_year,
            order_month,
            country,
            COUNT(*) as total_orders,
            ROUND(SUM(amount), 2) as total_revenue,
            ROUND(AVG(amount), 2) as avg_order_value,
            COUNT(CASE WHEN status = 'completed' THEN 1 END) as completed_orders
        FROM clean_orders
        GROUP BY order_year, order_month, country
        ORDER BY order_year, order_month, total_revenue DESC
    """)

    print("\nMonthly Revenue Summary:")
    summary_df.show(truncate=False)

What does this SQL query do?

  • .createOrReplaceTempView("clean_orders") registers the DataFrame so you can reference it as a SQL table named clean_orders.
  • The query groups orders by year, month, and country to calculate total orders, revenue, average order value, and completed order count.
  • COUNT(CASE WHEN status = 'completed' THEN 1 END) counts only rows where the status is completed, ignoring shipped, pending, and other statuses.
  • .show(truncate=False) prints the full summary table without cutting off wide columns.
Write the summary and run the complete pipeline

The summary DataFrame contains your aggregated business insights. You'll save it to its own Parquet file, then read it back to confirm everything wrote correctly.

  • Add the following code directly below the summary_df.show(truncate=False) line to finish the load() function:
    # Write summary to Parquet
    summary_df.write.mode("overwrite").parquet(output_summary_path)
    print(f"Summary written to: {output_summary_path}")

    # Verify by reading back
    verification_df = spark.read.parquet(output_summary_path)
    print(f"\nVerification - Summary rows: {verification_df.count()}")

    return summary_df

What does this code do?

  • summary_df.write.mode("overwrite").parquet() saves the aggregated results to data/processed/summary/.
  • The verification step reads the Parquet file back and prints the row count. This confirms the write succeeded and the data is readable.
  • return summary_df passes the summary back to main() in case you want to inspect it further.

Now you need to update main() to call your new load() function and print a completion message.

  • In the main() function, find this line inside the try block:
        # Transform
        clean_df = transform(raw_df)
  • Add the following code directly below it, still inside the try block:
        # Load
        load(spark, clean_df, output_orders_path, output_summary_path)

        print("\n" + "#" * 50)
        print("  PIPELINE COMPLETED SUCCESSFULLY")
        print("#" * 50)

What does this addition do?

This connects your load() function into the pipeline orchestration. The main() function now runs all three ETL phases (extract, transform, load) in sequence, then prints a success banner.

  • Save src/etl_pipeline.py.
  • Run the full pipeline by running this command in your terminal:
python src/etl_pipeline.py
  • You should see all three phases print in sequence: EXTRACT PHASE, TRANSFORM PHASE, and LOAD PHASE, followed by a Monthly Revenue Summary table and the PIPELINE COMPLETED SUCCESSFULLY banner.

Seeing an error when running the pipeline?

  • If you see IndentationError, check that the code inside load() is indented with 4 spaces, and the code inside the try block uses 8 spaces.
  • If you see NameError: name 'load' is not defined, make sure the load() function is defined outside of main(), between transform() and main().
  • If you see a file path error, confirm you're running the command from the ecommerce_etl project folder (not from inside src/).

Help me debug my PySpark pipeline error

  • Verify the output files were created by running:
ls data/processed/
  • You should see orders and summary directories listed.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete src/etl_pipeline.py file. Compare yours against this:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, to_date, year, month, when, coalesce, lit, sum as spark_sum, count, avg
import os


def create_spark_session():
    """Create and return a SparkSession."""
    return SparkSession.builder \
        .appName("Ecommerce_ETL_Pipeline") \
        .config("spark.sql.adaptive.enabled", "true") \
        .getOrCreate()


def extract(spark, input_path):
    """Extract: Read raw CSV data into a DataFrame."""
    print("=" * 50)
    print("EXTRACT PHASE")
    print("=" * 50)

    df = spark.read.csv(input_path, header=True, inferSchema=True)

    print(f"\nLoaded {df.count()} rows from {input_path}")
    print("\nSchema:")
    df.printSchema()
    print("\nSample data:")
    df.show(5, truncate=False)

    return df


def transform(df):
    """Transform: Clean and enrich the data."""
    print("\n" + "=" * 50)
    print("TRANSFORM PHASE")
    print("=" * 50)

    initial_count = df.count()

    # Drop rows with null order_id or customer_id
    df_clean = df.dropna(subset=["order_id", "customer_id"])

    # Filter out negative amounts
    df_clean = df_clean.filter(col("amount") > 0)

    cleaned_count = df_clean.count()
    print(f"\nRows removed during cleaning: {initial_count - cleaned_count}")

    # Standardize status to lowercase
    df_clean = df_clean.withColumn("status", lower(col("status")))

    # Parse dates - handle multiple formats using coalesce
    df_clean = df_clean.withColumn(
        "order_date_parsed",
        coalesce(
            to_date(col("order_date"), "yyyy-MM-dd"),
            to_date(col("order_date"), "MM/dd/yyyy")
        )
    )

    # Derive year and month columns
    df_clean = df_clean.withColumn("order_year", year(col("order_date_parsed")))
    df_clean = df_clean.withColumn("order_month", month(col("order_date_parsed")))

    # Add amount category
    df_clean = df_clean.withColumn(
        "amount_category",
        when(col("amount") < 100, "small")
        .when(col("amount") < 500, "medium")
        .otherwise("large")
    )

    # Drop the original order_date and rename parsed date
    df_clean = df_clean.drop("order_date") \
        .withColumnRenamed("order_date_parsed", "order_date")

    print(f"\nRows after cleaning: {cleaned_count}")
    print("\nCleaned data sample:")
    df_clean.show(5, truncate=False)

    return df_clean


def load(spark, df_clean, output_orders_path, output_summary_path):
    """Load: Write cleaned data and aggregated summary to Parquet."""
    print("\n" + "=" * 50)
    print("LOAD PHASE")
    print("=" * 50)

    # Write partitioned Parquet
    df_clean.write \
        .mode("overwrite") \
        .partitionBy("order_year", "order_month") \
        .parquet(output_orders_path)

    print(f"\nCleaned orders written to: {output_orders_path}")

    # Create aggregated summary using Spark SQL
    df_clean.createOrReplaceTempView("clean_orders")

    summary_df = spark.sql("""
        SELECT
            order_year,
            order_month,
            country,
            COUNT(*) as total_orders,
            ROUND(SUM(amount), 2) as total_revenue,
            ROUND(AVG(amount), 2) as avg_order_value,
            COUNT(CASE WHEN status = 'completed' THEN 1 END) as completed_orders
        FROM clean_orders
        GROUP BY order_year, order_month, country
        ORDER BY order_year, order_month, total_revenue DESC
    """)

    print("\nMonthly Revenue Summary:")
    summary_df.show(truncate=False)

    # Write summary to Parquet
    summary_df.write.mode("overwrite").parquet(output_summary_path)
    print(f"Summary written to: {output_summary_path}")

    # Verify by reading back
    verification_df = spark.read.parquet(output_summary_path)
    print(f"\nVerification - Summary rows: {verification_df.count()}")

    return summary_df


def main():
    """Main ETL pipeline orchestration."""
    print("\n" + "#" * 50)
    print("  ECOMMERCE ETL PIPELINE - STARTING")
    print("#" * 50)

    # Paths
    input_path = os.path.join("data", "raw", "orders.csv")
    output_orders_path = os.path.join("data", "processed", "orders")
    output_summary_path = os.path.join("data", "processed", "summary")

    # Create SparkSession
    spark = create_spark_session()

    try:
        # Extract
        raw_df = extract(spark, input_path)

        # Transform
        clean_df = transform(raw_df)

        # Load
        load(spark, clean_df, output_orders_path, output_summary_path)

        print("\n" + "#" * 50)
        print("  PIPELINE COMPLETED SUCCESSFULLY")
        print("#" * 50)

    except Exception as e:
        print(f"\nPIPELINE FAILED: {e}")
        raise

    finally:
        spark.stop()
        print("\nSpark session closed.")


if __name__ == "__main__":
    main()

Secret mission

Add a Data Quality Report

Your pipeline processes data, but how do you know it's healthy? Production pipelines include monitoring that tracks how much data was cleaned, flags anomalies, and times each stage. Build a data quality layer that generates a JSON report after every run.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources for future experimentation, or delete them to free up disk space.

Resources you used:

  • ecommerce_etl project folder on your Desktop (source code, virtual environment, raw data, and processed output).
  • Java 17 (Microsoft Build of OpenJDK) installed system-wide via winget.

Keep everything running

No action needed. Choose this if you plan to experiment further with PySpark or extend the pipeline with new data sources.

  • Your project folder stays on your Desktop at ecommerce_etl.
  • Java 17 remains installed on your system. Other Java-based tools (Kafka, Hadoop, Elasticsearch) can reuse it.
  • The virtual environment stays inside your project folder. Reactivate it any time to run the pipeline again.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files stay on disk for later.

  • Deactivate the virtual environment by running:
  • This returns your terminal to the system Python. Your ecommerce_etl folder and all processed data remain on disk.
deactivate

When you're ready to resume, navigate back to your project folder and reactivate the environment.

Delete - I don't want to use this again

Remove all project files and optionally uninstall Java 17.

  • Delete the entire project folder by running this command in Command Prompt:
rmdir /s /q %USERPROFILE%\Desktop\ecommerce_etl
  • Or in PowerShell:
Remove-Item -Recurse -Force ~/Desktop/ecommerce_etl

This removes your source code, virtual environment, raw CSV data, and all processed Parquet output.

  • To uninstall Java 17, run:
winget uninstall Microsoft.OpenJDK.17

Should I uninstall Java?

Only uninstall Java if no other applications on your system depend on it. Tools like Elasticsearch, Kafka, and Android Studio require Java. If you're unsure, leave it installed.

Nice Work!

Nice Work!

You just built a production-style ETL pipeline with PySpark. This is the exact same pattern used at companies running data platforms on AWS Glue, Databricks, and EMR.

You've learned how to:

  • Built a PySpark ETL pipeline that processes raw CSV data through extract, transform, and load stages.
  • Cleaned messy data using DataFrame transformations: filtering nulls, standardizing text, parsing mixed date formats, and deriving new columns.
  • Wrote Spark SQL queries to generate business insights, then output clean Parquet files partitioned and ready for analytics tools.
  • Added a data quality monitoring layer that flags anomalies, counts nulls, tracks timing per stage, and generates a JSON health report.

Ready to quiz yourself?