Sales Analytics Pipeline with PySpark

Build a local ETL pipeline that loads, cleans, joins, and queries sales data.

Introduction

30 Second Summary

Every online order you place generates a row of data. Multiply that by millions of customers and you have the raw material that data engineers transform into business insights every day.

In this project, you will build a PySpark pipeline that processes raw e-commerce transaction data from messy CSV files into clean, queryable analytics output saved as optimized Parquet files.

What You'll Build

You run a single Python script and watch it load raw CSV files, clean dirty data, join two datasets together, compute revenue metrics, and save everything as Parquet files ready for a data warehouse.

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

  • A working local PySpark environment where you type one command and see your entire pipeline run from raw data to finished output.
  • A demo-ready "before and after" showing how schema inference breaks on messy data and how explicit schemas fix it. This is the single most common PySpark interview gotcha.
  • Business analytics (revenue by category, top products, customer spending) queryable through both the DataFrame API and Spark SQL. Interviewers test both and expect you to switch between them.
  • Secret Mission: Add window functions to rank products within categories and compute running customer spending totals.

Are there any prerequisites?

You should be comfortable with Python basics (functions, lists, dictionaries). No prior PySpark or Java experience is required.

Before We Start

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

This project builds a complete PySpark sales analytics pipeline from scratch. The patterns you'll practice (schema definition, joins, aggregations, Spark SQL) are the exact patterns tested in data engineering interviews.

Install Java 17 and PySpark

PySpark runs on the Java Virtual Machine (JVM). Without Java installed, any attempt to start a SparkSession crashes immediately.

In this step, you'll install Java 17, set up PySpark in a virtual environment, and verify everything works by launching your first SparkSession.

In this step, get ready to:
  • Install Eclipse Temurin JDK 17 on Windows.
  • Install PySpark 4.2.0 in a Python virtual environment.
  • Create the project structure and run your first SparkSession.
Install Eclipse Temurin JDK 17 on Windows

PySpark 4.2.0 requires Java 17, 21, or 25. You'll install Eclipse Temurin JDK 17, a free and open-source distribution of Java.

  • Check if Java is already installed by opening a terminal and running:
java -version

✔️ I see version 17

You already have Java 17 installed. You're ready to move on to installing PySpark.

ⓧ I see a different version

You have Java installed, but PySpark 4.2.0 needs version 17, 21, or 25. Install Temurin JDK 17 to get the correct version.

  • Download the Windows x64 .msi installer from the Adoptium Temurin releases page.
  • Run the downloaded .msi file to launch the installer.
  • Accept the license agreement.
  • On the Custom Setup screen, click the directory tree icon next to Set JAVA_HOME variable and select Will be installed on local hard drive.
  • Confirm that Add to PATH is already checked (it is enabled by default).
  • Click Next, then click Install.
  • After installation completes, click Finish.

Why a new terminal?

Environment variables like JAVA_HOME and PATH are loaded when a terminal opens. Any terminal that was open before the install will not see the new Java.

  • Close your current terminal, open a new one, and verify by running:
java -version

You should see output starting with openjdk version "17 followed by the build number.

Still seeing the old version?

Make sure you opened a completely new terminal after the install finished. If the old version persists, restart your computer to force Windows to reload all environment variables.

help me fix my Java version after installing Temurin JDK 17 on Windows

ⓧ Command not found

Java is not installed on your machine. Install Temurin JDK 17 now.

  • Download the Windows x64 .msi installer from the Adoptium Temurin releases page.
  • Run the downloaded .msi file to launch the installer.
  • Accept the license agreement.
  • On the Custom Setup screen, click the directory tree icon next to Set JAVA_HOME variable and select Will be installed on local hard drive.
  • Confirm that Add to PATH is already checked (it is enabled by default).
  • Click Next, then click Install.
  • After installation completes, click Finish.
  • Open a new terminal and verify by running:
java -version

You should see output starting with openjdk version "17 followed by the build number.

Command still not found?

Make sure you opened a completely new terminal after the install finished. If it still does not work, restart your computer to force Windows to reload environment variables.

help me troubleshoot Java not being found after installing Temurin

Why Temurin over Oracle JDK?

Eclipse Temurin is free for all uses with no licensing concerns. Oracle JDK changed its license terms, so Temurin is the safer default for local development.

Install PySpark in a virtual environment

A virtual environment keeps your project's Python packages isolated from your system Python. You'll create one now and install PySpark inside it.

  • Open VS Code by pressing the Windows key, typing VS Code, and pressing Enter.
  • Open the integrated terminal by pressing Ctrl+` (backtick key, usually above Tab).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir sales-pipeline
cd sales-pipeline
  • Create a virtual environment by running:
python -m venv venv
  • Confirm the virtual environment was created by running:
ls venv

You should see folders like Include, Lib, and Scripts listed in the output.

  • Activate the virtual environment by running:
venv\Scripts\activate

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

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

What does this install?

This downloads PySpark 4.2.0 and all its dependencies into your virtual environment. The install may take a minute since PySpark bundles the Spark engine.

  • Verify PySpark installed correctly by running:
python -c "import pyspark; print(pyspark.__version__)"

You should see 4.2.0 printed in the terminal.

Seeing an error during pip install?

Make sure your virtual environment is active (you see (venv) in your prompt). If pip cannot find the package, check your internet connection and try again.

help me fix my PySpark installation

Create the project structure

Now that your tools are installed, you'll create the files your pipeline needs. The project starts with one Python script and two CSV data files.

  • Open the sales-pipeline folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the sales-pipeline folder, and click Select Folder.

Why open the folder in VS Code?

Opening a folder (not just a file) gives VS Code full project context. The Explorer sidebar shows all your files, and the integrated terminal opens inside the project directory automatically.

  • If VS Code opened a new window, open the terminal again with Ctrl+` and reactivate your virtual environment:
venv\Scripts\activate
  • Create the data directory and empty CSV placeholder files by running:
mkdir data
  • Create the two empty CSV files by running these commands:
New-Item data\orders.csv -ItemType File
New-Item data\products.csv -ItemType File
  • Confirm the files exist by running:
ls data

You should see orders.csv and products.csv listed in the output.

  • Create the main pipeline file. In the VS Code Explorer sidebar, right-click in the empty area and select New File.
  • Name it pipeline.py and press Enter.
  • Paste the following code into pipeline.py:
from pyspark.sql import SparkSession

# Create a SparkSession that runs locally on all available cores
spark = SparkSession.builder.master("local[*]").appName("SalesAnalytics").getOrCreate()
print(f"Spark version: {spark.version}")
spark.stop()

What does this code do?

  • SparkSession.builder creates the entry point to all Spark functionality.
  • master("local[*]") tells Spark to run on your local machine using all CPU cores.
  • appName("SalesAnalytics") gives your application a name visible in logs.
  • spark.stop() shuts down the SparkSession cleanly when the script finishes.
  • Save pipeline.py by pressing Ctrl+S.

Before you run this, what do you think will happen? Will Spark start successfully, or will it fail because of a missing dependency?

  • Run your pipeline by executing this command in the terminal:
python pipeline.py

You should see Spark startup logs followed by a line that reads Spark version: and then a version number. This confirms Java, PySpark, and your SparkSession all work together.

Seeing a Java-related error?

If you see an error mentioning Java or JAVA_HOME, your terminal may not have picked up the new environment variables. Close VS Code entirely, reopen it, reactivate your venv with venv\Scripts\activate, and try again.

help me fix Java errors when running PySpark

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and that your terminal shows the Spark version output.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py at this point:

from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").appName("SalesAnalytics").getOrCreate()
print(f"Spark version: {spark.version}")
spark.stop()

Your environment is ready. Java, PySpark, and your project files are all in place. Next up, you'll load real data into PySpark and discover why schema inference causes problems in production pipelines.

Load Data and Discover the Schema Inference Problem

Your Spark environment is verified and ready. Now you need actual data to process and a reason to care about how PySpark reads it.

Schema inference is the most common gotcha in PySpark interviews. When PySpark reads a CSV, it scans the file and guesses column types. If any value in a column breaks the numeric pattern (like "N/A" mixed in with numbers), Spark falls back to string. In this step, you will feel that problem firsthand.

In this step, get ready to:
  • Populate CSV data files with sample e-commerce data.
  • Load data with schema inference and observe wrong column types.
  • Attempt a calculation that fails due to the type errors.
Create sample CSV data files

Real-world data is messy. Your sample files will include intentional problems: an "N/A" string where a number should be, and a missing date field. These imperfections are what trip up schema inference.

  • Open data/orders.csv in VS Code and paste the following order data:
order_id,customer_id,product_id,order_amount,order_date,quantity
1,101,1,29.99,2024-01-15,2
2,102,2,49.99,2024-01-16,1
3,101,3,15.50,2024-01-17,4
4,103,1,29.99,2024-01-18,1
5,104,4,99.99,2024-01-19,2
6,102,5,N/A,2024-01-20,3
7,101,2,49.99,2024-01-21,1
8,105,3,15.50,2024-01-22,6
9,103,4,99.99,2024-01-23,2
10,104,1,29.99,2024-01-24,3
11,101,5,75.00,2024-01-25,1
12,102,1,29.99,,2
13,105,2,49.99,2024-01-27,4
14,103,3,15.50,2024-01-28,2
15,104,5,75.00,2024-01-29,1

What makes this data messy?

Row 6 has N/A in the order_amount column instead of a number. Row 12 has a completely empty order_date field. These are the exact kinds of dirty data that break schema inference in production.

  • Save data/orders.csv.
  • Open data/products.csv and paste the following product catalog:
product_id,product_name,category,unit_price
1,Wireless Mouse,Electronics,29.99
2,Mechanical Keyboard,Electronics,49.99
3,Notebook Pack,Office Supplies,15.50
4,Standing Desk,Furniture,99.99
5,USB-C Hub,Electronics,75.00
6,Desk Lamp,Furniture,35.00
7,Sticky Notes,Office Supplies,5.99
8,Monitor Stand,Furniture,45.00
  • Save data/products.csv.
  • Confirm both files saved by running:
ls data/

You should see orders.csv and products.csv listed in the output.

Not seeing both files?

Make sure you saved both files after pasting. Check that you are running ls data/ from your project folder (the one containing the data directory).

Help me find my CSV files

Load with schema inference and inspect the problem

PySpark's inferSchema=True option scans your CSV file and guesses what type each column should be. When it encounters "N/A" in a column alongside real numbers, it cannot parse it as a number and falls back to labeling the entire column as a string.

  • Replace the entire contents of pipeline.py with the following code that loads and inspects the orders data:
from pyspark.sql import SparkSession

def main():
    # Create SparkSession
    spark = (
        SparkSession.builder
        .master("local[*]")
        .appName("SalesAnalytics")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("WARN")
    print(f"Spark version: {spark.version}")

    # Load orders CSV with schema inference
    print("\n=== Loading with schema inference ===")
    orders_inferred = spark.read.csv("data/orders.csv", header=True, inferSchema=True)
    print("Inferred schema:")
    orders_inferred.printSchema()
    orders_inferred.show()

    spark.stop()

if __name__ == "__main__":
    main()

What does this code do?

  • spark.read.csv("data/orders.csv", header=True, inferSchema=True) reads the CSV and tells Spark to guess column types by scanning the data.
  • printSchema() displays the type Spark chose for each column.
  • show() prints the first 20 rows so you can see the actual values.
  • setLogLevel("WARN") reduces noisy INFO logs so you can focus on the output that matters.
  • Save pipeline.py and run it:
python pipeline.py

Look at the schema output. You should see order_amount listed as string type instead of double. The single "N/A" value in row 6 forced Spark to treat the entire column as text.

Why did Spark choose string?

Spark's inference engine tries to parse each value in a column as Integer, then Long, then Double, then falls back to String. The "N/A" in row 6 cannot be parsed as any numeric type. Since not every value is numeric, Spark picks the safest common type: string.

The order_date column may also appear as string instead of date. Inference struggles with date formats too.

Seeing a Java error instead of schema output?

If you see a JAVA_HOME error, make sure you opened a new terminal after installing Java. The environment variables only apply to new sessions.

If you see a file-not-found error, confirm you are running the command from your project folder (the one containing the data directory).

Help me debug my PySpark Java error

Attempt a sum and see it fail

The real pain of wrong types shows up the moment you try to do math. A string column cannot be summed. This is the problem every production pipeline must solve.

  • In pipeline.py, find the import line at the top:
from pyspark.sql import SparkSession
  • Add this import directly below it:
from pyspark.sql.functions import sum
  • Now find the line spark.stop() near the bottom of your main() function. Add this block directly above it:
    # Try to sum order_amount - fails because column is string
    print("Attempting sum with inferred schema:")
    try:
        orders_inferred.select(sum("order_amount")).show()
    except Exception as e:
        print(f"Error: {e}")

What does this code do?

  • sum("order_amount") attempts to add up every value in the order_amount column.
  • The try/except block catches the error so the script does not crash. You will see either an error message or a null result, because you cannot sum strings.
  • Save pipeline.py.

Before you run this, what do you think will happen when PySpark tries to sum a column that it labeled as string? Will it produce a number, return null, or throw an error?

  • Run the pipeline:
python pipeline.py

You will see the sum attempt either throw an error or return null. Either way, the calculation fails. This is the direct consequence of schema inference labeling order_amount as a string.

Why this matters in interviews

Interviewers expect you to know that production pipelines never rely on inferSchema=True. Inference scans the file twice (once to guess types, once to read data), slowing performance on large files. It also guesses wrong when messy data is involved.

The fix is explicit schemas using StructType. You will apply that fix in the next step.

Sum returned a number instead of failing?

Depending on your Spark version's exact behavior, sum on a string column may return null instead of throwing an exception. A null result is still a failure. The correct total should be a real number, not null.

Explain why my PySpark sum returned null

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is the complete pipeline.py at the end of this step:

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum

def main():
    spark = (
        SparkSession.builder
        .master("local[*]")
        .appName("SalesAnalytics")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("WARN")
    print(f"Spark version: {spark.version}")

    print("\n=== Loading with schema inference ===")
    orders_inferred = spark.read.csv("data/orders.csv", header=True, inferSchema=True)
    print("Inferred schema:")
    orders_inferred.printSchema()
    orders_inferred.show()

    print("Attempting sum with inferred schema:")
    try:
        orders_inferred.select(sum("order_amount")).show()
    except Exception as e:
        print(f"Error: {e}")

    spark.stop()

if __name__ == "__main__":
    main()

You have just experienced the schema inference problem firsthand. Your order_amount column is stuck as a string, and basic math fails. Next up, you will fix this with explicit schemas and clean the data into a usable state.

Fix the Schema and Clean the Data

In the previous step, you saw schema inference go wrong. PySpark guessed order_amount was a string because of the "N/A" value, and your sum() call failed.

The fix is to define an explicit schema using StructType. This tells PySpark exactly what type each column should be. Values that don't match (like "N/A" for a Double column) become null instead of corrupting the entire column's type. Every production pipeline uses this pattern.

In this step, get ready to:
  • Define explicit StructType schemas for your CSV files.
  • Clean dirty data by filtering nulls and enriching with a computed column.
  • Load the products dataset with its own explicit schema.
Define explicit schemas using StructType

A StructType is a list of StructField objects that map each column to a name, a data type, and whether it allows nulls. When you pass this schema to spark.read.csv(), PySpark skips inference entirely and uses your types.

  • In pipeline.py, find the import lines at the top of the file:
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum
  • Replace both import lines with these three imports:
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, DateType
)
from pyspark.sql.functions import col, sum, when

What are these new imports?

You are importing three groups of tools. pyspark.sql.types gives you the building blocks for schema definitions. col lets you reference columns by name in filter expressions. when lets you write conditional logic (like an if/else) for computed columns.

  • Now find the line that says spark.stop() near the bottom of your main() function. Add the following code above it to define the orders schema and reload the data:
    print("\n=== Loading with explicit schema ===")
    orders_schema = StructType([
        StructField("order_id", IntegerType(), True),
        StructField("customer_id", IntegerType(), True),
        StructField("product_id", IntegerType(), True),
        StructField("order_amount", DoubleType(), True),
        StructField("order_date", DateType(), True),
        StructField("quantity", IntegerType(), True),
    ])

    orders_raw = spark.read.csv("data/orders.csv", header=True, schema=orders_schema)
    print("Explicit schema:")
    orders_raw.printSchema()

What does this code do?

Each StructField defines one column: its name, its type, and whether it can be null (True means yes). By specifying DoubleType() for order_amount, PySpark tries to parse each value as a number. The "N/A" string can't be parsed, so it becomes null instead of forcing the whole column to string.

Similarly, DateType() for order_date means PySpark parses date strings into proper date objects. Empty date cells become null.

  • Save pipeline.py and run the pipeline to see the fixed schema:
python pipeline.py

In the new "Loading with explicit schema" section, you should see order_amount: double and order_date: date in the printed schema. The type problem is fixed.

Seeing an import error?

Make sure your import statement matches exactly. StructType and StructField must both come from pyspark.sql.types. Check that you have the opening and closing parentheses around the multi-line import.

Help me fix this import error

Clean the data using DataFrame operations

Your schema is correct now, but the DataFrame still has rows with null values where "N/A" and empty strings used to be. You need to filter those out and add a useful computed column.

  • In pipeline.py, add the following code directly below the orders_raw.printSchema() line you just added:
    # Clean: remove rows with null order_amount or order_date
    orders_clean = orders_raw.filter(
        col("order_amount").isNotNull() & col("order_date").isNotNull()
    )

    # Enrich: add order_size column
    orders_df = orders_clean.withColumn(
        "order_size",
        when(col("quantity") < 3, "small")
        .when(col("quantity") <= 5, "medium")
        .otherwise("large")
    )
    print("Clean and enriched orders:")
    orders_df.show()

What does this code do?

The filter() call removes rows where order_amount or order_date is null. This drops the two messy rows (the "N/A" row and the empty-date row) in one step.

The withColumn call creates a new column called order_size based on conditional logic. If quantity is less than 3, it labels the order "small". If quantity is 3 to 5, it labels it "medium". Otherwise, "large". This is how you enrich data with business logic in PySpark.

  • Save pipeline.py and run the pipeline again:
python pipeline.py

You should see 13 rows in the "Clean and enriched orders" table (down from 15). Each row now has an order_size column showing "small", "medium", or "large".

Still seeing 15 rows?

If null rows are not being filtered, check that you typed isNotNull() with the parentheses. Also verify the & operator is between the two conditions. PySpark uses & for AND (not the word and).

Help me debug my PySpark filter

Load and define schema for products

Your orders data is clean. Now you need to load the products dataset with the same explicit-schema approach so both DataFrames are ready for joining in the next step.

  • In pipeline.py, add the following code below the orders_df.show() line (and still above spark.stop()):
    products_schema = StructType([
        StructField("product_id", IntegerType(), True),
        StructField("product_name", StringType(), True),
        StructField("category", StringType(), True),
        StructField("unit_price", DoubleType(), True),
    ])

    products_df = spark.read.csv("data/products.csv", header=True, schema=products_schema)
    print("Products:")
    products_df.printSchema()
    products_df.show()

What does this code do?

This follows the same pattern as your orders schema. You define the exact types for each column, then pass the schema to spark.read.csv(). Because the products data is already clean (no messy values), you don't need to filter anything here.

  • Save pipeline.py.

Before you run the pipeline, what do you expect to see for the products schema types? Will unit_price show as double or string?

  • Run the pipeline to check your prediction:
python pipeline.py

You should see the products schema with unit_price: double and all 8 product rows displayed. Both DataFrames now have correct types guaranteed by your explicit schemas.

Seeing a file not found error for products.csv?

Make sure your data/products.csv file exists in the data folder inside your project directory. The path is relative to where you run python pipeline.py from.

Help me fix this file path error

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and your terminal shows all three outputs: the inferred schema (still broken), the explicit schema (fixed), and the products table.

ⓧ I'd like to double check the full code

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

  • Three import lines at the top: SparkSession, the types import (StructType, StructField, etc.), and the functions import (col, sum, when).
  • The main() function starts by creating a SparkSession and loading with inferSchema=True (the broken version from Step 2).
  • A try/except block that attempts sum("order_amount") on the inferred DataFrame.
  • The explicit schema section: orders_schema definition, orders_raw loaded with schema, printSchema().
  • The cleaning section: orders_clean filtering nulls, orders_df adding order_size, then show().
  • The products section: products_schema definition, products_df loaded with schema, printSchema() and show().
  • Finally, spark.stop() at the end of main().

Your data is clean, typed correctly, and enriched with a computed column. Next up, you will join these two DataFrames together and run aggregation queries to extract real business insights.

Join, Aggregate, and Query with SQL

Your pipeline now loads both CSVs with explicit schemas, filters out dirty rows, and enriches orders with a computed order_size column. You have two clean DataFrames ready to combine.

Joins and aggregations are the bread and butter of data engineering interviews. Interviewers expect you to combine datasets, compute business metrics, and show you can use both the DataFrame API and Spark SQL interchangeably. In this step, you will do exactly that.

In this step, get ready to:
  • Join orders with products using an inner join.
  • Compute revenue metrics grouped by category.
  • Query your data using Spark SQL temporary views.
Join orders with products

An inner join matches rows from both DataFrames where the join key exists in both. Rows in orders that reference a product_id present in products will combine into a single wide row.

  • In pipeline.py, find the spark.stop() line at the bottom of your main() function.
  • Add the following code above spark.stop() to join the two DataFrames and display the result:
    print("\n=== Join orders with products ===")
    joined_df = orders_df.join(products_df, on="product_id", how="inner")
    joined_df.select(
        "order_id", "customer_id", "product_name", "category",
        "order_amount", "order_date", "quantity", "order_size"
    ).show()

What does this code do?

  • orders_df.join(products_df, on="product_id", how="inner") matches each order to its product row using the shared product_id column.
  • The .select() picks only the columns you care about, keeping the output readable.
  • An inner join drops any order that references a product not in products.csv. In interviews, you would explain when a left join is more appropriate.
  • Save pipeline.py.
  • Run the pipeline to see the joined output:
python pipeline.py
  • You should see a table titled === Join orders with products === showing rows that now include product_name and category alongside each order.

Seeing an empty table or AnalysisException?

  • Check that both DataFrames have a column named exactly product_id. If the CSV header has a trailing space, the join silently produces zero matches.
  • Make sure your new code is indented inside the main() function and sits above spark.stop().

Why is my PySpark join returning zero rows?

Compute aggregations with groupBy

Now that you have a single DataFrame with both order and product data, you can answer business questions like "which category generates the most revenue?" using groupBy and agg.

  • In pipeline.py, find the import line at the top of the file:
from pyspark.sql.functions import col, sum, when
  • Replace it with this updated import that adds avg and count:
from pyspark.sql.functions import col, sum, avg, count, when
  • Now add the aggregation code below your .show() call on the joined DataFrame (and still above spark.stop()):
    print("=== Revenue by Category ===")
    category_revenue_df = (
        joined_df
        .groupBy("category")
        .agg(
            sum("order_amount").alias("total_revenue"),
            avg("order_amount").alias("avg_order_value"),
            count("order_id").alias("total_orders"),
        )
        .orderBy(col("total_revenue").desc())
    )
    category_revenue_df.show()

What does this code do?

  • groupBy("category") splits the DataFrame into groups. Each unique category (Electronics, Furniture, Office Supplies) becomes its own bucket.
  • .agg() computes multiple aggregate functions at once: total revenue, average order value, and order count per category.
  • .alias() renames the output columns so they are readable instead of showing raw function names like sum(order_amount).
  • .orderBy(col("total_revenue").desc()) sorts categories from highest to lowest revenue.
  • Save pipeline.py and run it:
python pipeline.py
  • You should see a new table titled === Revenue by Category === showing three rows (one per category) with total_revenue, avg_order_value, and total_orders columns.

Seeing a NameError for avg or count?

  • Make sure you updated the import line at the top to include avg and count. The original import only had col, sum, when.
  • Double-check that you imported from pyspark.sql.functions, not Python's built-in sum or count.

Help me fix my PySpark import error.

Run equivalent queries with Spark SQL

Many data engineering teams use SQL as their primary query language. PySpark lets you register any DataFrame as a temporary view and query it with standard SQL syntax. This means you can switch between the DataFrame API and SQL depending on what reads more clearly.

  • In pipeline.py, add the following code below your category_revenue_df.show() line (and still above spark.stop()):
    print("=== Spark SQL: Top 3 Products by Revenue ===")
    joined_df.createOrReplaceTempView("sales")
    top_products = spark.sql(
        """
        SELECT product_name, SUM(order_amount) AS total_revenue
        FROM sales
        GROUP BY product_name
        ORDER BY total_revenue DESC
        LIMIT 3
        """
    )
    top_products.show()

What does this code do?

  • createOrReplaceTempView("sales") registers joined_df as a table named sales that you can query with SQL. The view lives only for this SparkSession.
  • spark.sql() executes standard SQL against that view. The query groups by product name, sums revenue, sorts descending, and takes the top 3.
  • This produces the same result you could get with the DataFrame API. Interviewers test both to see which you reach for and whether you can translate between them.
  • Save pipeline.py.

Before you run the pipeline, predict: how many separate output tables will you see in the terminal now?

  • Run the pipeline:
python pipeline.py

You should see three new outputs in the terminal: the joined DataFrame with product names, the revenue-by-category aggregation table, and a Spark SQL result showing the top 3 products by total revenue.

SQL query returning an error about the sales table?

  • Make sure createOrReplaceTempView("sales") is called on joined_df before the spark.sql() call. The view must exist before you query it.
  • Check that the table name in your SQL string (FROM sales) matches the name you registered ("sales").

Help me debug my Spark SQL temporary view error.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py with all three new sections (join, aggregation, SQL query).

ⓧ I'd like to double check the full code

Scroll up and confirm these sections are present in your pipeline.py, in this order:

  • The updated import line with col, sum, avg, count, when.
  • All existing code (SparkSession, schema inference demo, explicit schema, clean/enrich, products load).
  • The join block: orders_df.join(products_df, on="product_id", how="inner") with .select(...).show().
  • The aggregation block: joined_df.groupBy("category").agg(...) with .orderBy(col("total_revenue").desc()).
  • The Spark SQL block: createOrReplaceTempView("sales") followed by spark.sql(...).
  • spark.stop() at the very end of main().

Your pipeline now combines two datasets, computes aggregations, and runs SQL queries. Next up, you will save these results to Parquet format and prove why it is the production standard for data storage.

Save to Parquet and Verify the Output

Your pipeline now loads, cleans, joins, and aggregates data. The last piece is saving the results in a format that production systems actually use.

CSV is fine as an input format, but it has a critical flaw: it carries no schema information. Every time you read a CSV, you must either re-infer types (slow, error-prone) or redefine a schema. Parquet solves this by embedding the schema directly into the file. In this step, you will write your results to Parquet, read them back, and prove the schema survives the round-trip.

In this step, get ready to:
  • Write your aggregation and joined DataFrames to Parquet files.
  • Read back from Parquet and confirm the schema is preserved without inference.
  • Stop the SparkSession and complete the end-to-end pipeline.
Write aggregation results to Parquet

Right now your pipeline prints results to the terminal and then shuts down. Those results vanish as soon as the script ends. Production pipelines persist their output to durable storage.

  • In pipeline.py, find the spark.stop() line at the bottom of your main() function.
  • Replace spark.stop() with the following code to save both DataFrames to Parquet:
    # --- Step 5: Save to Parquet ---
    print("\n=== Saving to Parquet ===")
    category_revenue_df.write.mode("overwrite").parquet("output/sales_by_category.parquet")
    joined_df.write.mode("overwrite").parquet("output/enriched_sales.parquet")

What does this code do?

.write.mode("overwrite").parquet(path) saves the DataFrame as a folder of Parquet part-files at the given path. The overwrite mode means re-running the script replaces old output instead of crashing with a "path already exists" error.

You are saving two outputs: the revenue-by-category summary and the full enriched sales dataset. Both go into an output/ folder that Spark creates automatically.

  • Save pipeline.py and run the pipeline:
python pipeline.py
  • Confirm the output folder was created by running:
ls output

You should see two folders listed: enriched_sales.parquet and sales_by_category.parquet.

Don't see the output folder?

  • Make sure you removed the old spark.stop() line. If it still exists above your new code, Spark shuts down before writing.
  • Check your terminal is in the project folder (the same folder that contains pipeline.py).
  • Look for error messages in the terminal output. A NameError means a variable name is misspelled.

Why isn't my pipeline creating the output folder?

Read back from Parquet and verify the round-trip

The real power of Parquet is that it carries its schema with it. When you read a Parquet file, you never need to define a schema or infer types. The file already knows its own column names and data types.

  • In pipeline.py, add the following code directly below the two .write.mode("overwrite").parquet(...) lines you just added:
    # Read back and verify
    print("Reading back from Parquet:")
    parquet_df = spark.read.parquet("output/sales_by_category.parquet")
    parquet_df.printSchema()
    parquet_df.show()
    print("CSV required schema inference. Parquet embeds the schema. This is why production pipelines save to Parquet.")

What does this code do?

spark.read.parquet(path) loads the Parquet file back into a DataFrame. Notice there is no schema= argument and no inferSchema=True. The schema is already embedded in the Parquet metadata.

printSchema() will show the exact same types you defined earlier (double for revenue, long for counts) without any guessing.

  • Save pipeline.py and run the pipeline again:
python pipeline.py

Scroll to the end of the output. You should see the schema printed with correct types and the data displayed below it.

Seeing a path not found error?

  • Make sure the read-back code comes AFTER the write code. If you placed it above the .write.mode("overwrite").parquet(...) lines, the files do not exist yet when Spark tries to read them.
  • Confirm the path string matches exactly: "output/sales_by_category.parquet".

Why is my Parquet read failing with a path not found error?

Stop the SparkSession and complete the pipeline

Every PySpark script should cleanly shut down its SparkSession when finished. This releases the JVM resources and closes any open connections.

  • Add the following code at the very end of your main() function, below the read-back code you just added:
    # Stop
    spark.stop()
    print("\nPipeline complete. Output saved to output/ folder.")

What does this code do?

spark.stop() shuts down the SparkSession and its underlying JVM process. The print statement confirms the full ETL pipeline ran successfully from start to finish.

Before you run the final version, what do you expect to see at the end of the output?

  • Save pipeline.py and run the complete pipeline one final time:
python pipeline.py

At the bottom of the output, you should see the Parquet schema with total_revenue: double and total_orders: long, followed by the data table and the message confirming the pipeline is complete.

Why Parquet instead of CSV?

You just proved the core reason: Parquet embeds its schema. When you read the Parquet file back, PySpark already knew every column name and type without any schema definition or inference step.

In production, Parquet also enables column pruning (reading only the columns you need) and predicate pushdown (skipping rows that do not match your filter). These features make Parquet dramatically faster than CSV for analytical workloads.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that your output/ folder contains both sales_by_category.parquet and enriched_sales.parquet.

ⓧ I'd like to double check the full code

Scroll back through this step and confirm your pipeline.py has these sections at the bottom of main(), in this order:

  • The Parquet write block (# --- Step 5: Save to Parquet --- comment, print statement, two .write.mode("overwrite").parquet(...) calls).
  • The read-back block (# Read back and verify comment, spark.read.parquet(...), printSchema(), show(), comparison print).
  • The stop block (# Stop comment, spark.stop(), pipeline complete print).

Make sure the old standalone spark.stop() line is gone. Only the one inside the stop block should remain.

Your complete ETL pipeline now loads raw CSV, cleans and enriches the data, joins datasets, computes aggregations, queries with SQL, and saves optimized Parquet output. Next up, you will take on the Secret Mission and add advanced window functions to rank products and track customer spending over time.

Secret mission

Add Window Functions for Advanced Analytics

Window functions are the number one advanced topic in data engineering interviews. They separate mid-level candidates from senior ones. In this mission, you will add row_number() to rank the top product per category and a running sum to track cumulative customer spending over time.

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 runs entirely locally, so there are no ongoing costs.

Resources you used:

  • Virtual environment folder (venv/) inside your project folder.
  • Parquet output files in the output/ folder.
  • Java JDK 17 (Eclipse Temurin) system installation.

Keep everything running

No action needed. Choose this if you plan to keep building PySpark pipelines or want to revisit this project later.

  • Your virtual environment, output files, and Java installation all remain in place.
  • To pick up where you left off, open a terminal in your project folder, activate the venv with venv\Scripts\activate, and run python pipeline.py.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. All files stay on disk so you can return anytime.

  • If your venv is currently active, deactivate it by running:
deactivate
  • Your venv/ folder, output/ files, and Java installation all remain on disk.
  • To resume later, reactivate the venv with venv\Scripts\activate and you are back in business.

Delete - I don't want to use this again

Remove all project files and optionally uninstall Java to fully clean up your machine.

  • If your venv is currently active, deactivate it first by running:
deactivate
  • Navigate one level above your project folder, then delete the entire project folder by running:
cd ..
rmdir /s /q sales-analytics

What does this remove?

This deletes your venv/ folder, pipeline.py, the data/ folder, and the output/ folder all at once.

  • To uninstall Java JDK 17, press the Windows key and type Add or Remove Programs.
  • Find Eclipse Temurin JDK with Hotspot 17 in the list.
  • Click the three-dot menu next to it and select Uninstall.
  • Only uninstall Java if you are sure you will not use PySpark or any other JVM-based tool again.

That's a Wrap!

That's a Wrap!

Nice work! You just built a complete sales analytics pipeline with PySpark, processing raw CSV data into production-ready Parquet output.

You've learned how to:

  • Install Java 17 and PySpark 4.2.0 on Windows and create a working local Spark environment from scratch.
  • Discover why schema inference fails on messy data and define explicit StructType schemas to guarantee correct data types.
  • Build a complete ETL pipeline that loads CSV, cleans dirty rows, joins datasets, computes aggregations with groupBy, queries with Spark SQL, and saves results to Parquet format.
  • Secret Mission: Implement window functions (row_number and running sum) for advanced analytics that separate mid-level from senior data engineering candidates.

Ready to quiz yourself?