Skip to main content

Lab 1 — Build a PySpark batch pipeline

Objective​

By the end of this lab you will have a runnable PySpark script, batch_pipeline.py, that reads a raw orders CSV, cleans and deduplicates the records, aggregates revenue and order count per category, and writes the result as Parquet partitioned by category. It is the ingest-clean-aggregate-write core of every batch job in this school, built end to end and verified against a known answer. It runs on your own machine in Spark local mode — a single-process Spark that behaves like a cluster of one — so it costs nothing.

Skills exercised:

  • Ingesting CSV with an explicit schema instead of trusting inference
  • Cleaning with narrow filters and deduplicating with a window function
  • Aggregating with groupBy and writing partitioned Parquet
  • Verifying a pipeline's output against an expected result

Prerequisites and setup​

Prerequisites: the three lessons in this module — Spark's execution model, DataFrame transformations and actions, and Partitions, shuffles, and performance. You should also have finished DE-103 Python for Data Engineering.

Setup: local mode needs a Java runtime (Spark runs on the JVM) and Python 3.10 or newer. Install a JDK and the pyspark package, then verify both. Create a working folder for the lab.

java -version
python3 -m venv .venv && source .venv/bin/activate
pip install "pyspark==3.5.1"
mkdir de202-batch-lab && cd de202-batch-lab

The java -version command should print a version line such as openjdk version "17.0.10". If Java is not found, install a JDK (version 11 or 17) before continuing; PySpark cannot start without one. The pip install line pulls in PySpark and its bundled Spark. Use python3 on macOS/Linux and python on Windows for the steps below.

note

Spark local mode starts a fresh single-process Spark each time you run the script. If you close your terminal, re-activate the virtualenv and cd de202-batch-lab before running again; your files are still there.

info

This lab pins pyspark==3.5.1 so the explain() output and Parquet behavior match what you see here. A newer Spark will still run the pipeline, but plan text and defaults can differ.

Step 1 — Create the raw orders and start a Spark session​

Goal: create the raw input file and confirm PySpark can start a local session.

Save this exactly as raw_orders.csv:

order_id,event_time,customer_id,category,amount,status
1001,2026-03-01T09:00:00,c-01,books,20.00,completed
1002,2026-03-01T09:05:00,c-02,electronics,300.00,completed
1003,2026-03-01T09:10:00,c-03,books,,completed
1004,2026-03-01T09:15:00,c-01,electronics,150.00,cancelled
1005,2026-03-01T09:20:00,c-04,home,45.50,completed
1002,2026-03-01T10:00:00,c-02,electronics,320.00,completed
1006,2026-03-01T09:30:00,c-05,home,-10.00,completed
1007,2026-03-01T09:35:00,c-06,books,15.00,completed
1008,2026-03-01T09:40:00,c-07,electronics,500.00,completed
1009,2026-03-01T09:45:00,c-08,home,60.00,completed
1010,2026-03-01T09:50:00,c-09,books,25.00,completed
1005,2026-03-01T08:00:00,c-04,home,40.00,completed

This file carries defects on purpose: 1003 has a missing amount, 1004 is cancelled, 1006 has a negative amount, and 1002 and 1005 each appear twice with different timestamps. Your pipeline will drop the bad rows and keep the latest version of each duplicate.

Now create batch_pipeline.py with just enough to start a session:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("de202-batch-pipeline")
    .master("local[2]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

print("spark version:", spark.version)

Checkpoint: run the script and confirm Spark starts.

python3 batch_pipeline.py

You should see spark version: 3.5.1 after a few seconds of startup logging. If you see JAVA_HOME is not set or Unable to locate a Java Runtime, install a JDK and re-run — see the Troubleshooting table.

Step 2 — Ingest with an explicit schema​

Goal: read the CSV into a DataFrame using a schema you declare, so a bad column never gets silently guessed.

Replace the print line in batch_pipeline.py with the schema and read:

from pyspark.sql.types import (
    StructType,
    StructField,
    StringType,
    DoubleType,
)

schema = StructType(
    [
        StructField("order_id", StringType(), nullable=False),
        StructField("event_time", StringType(), nullable=False),
        StructField("customer_id", StringType(), nullable=True),
        StructField("category", StringType(), nullable=False),
        StructField("amount", DoubleType(), nullable=True),
        StructField("status", StringType(), nullable=False),
    ]
)

raw = spark.read.csv("raw_orders.csv", header=True, schema=schema)
print("ingested rows:", raw.count())

Declaring the schema does two things at once: it skips Spark's inference pass over the file, and it turns the empty amount on row 1003 into a proper null Double rather than the string "".

Checkpoint: run it. You should see ingested rows: 12 — every raw row, defects included. If you see 11, your CSV is missing a line; re-save all twelve data rows from Step 1.

Step 3 — Clean with narrow filters​

Goal: keep only completed orders with a real, positive amount, using the narrow transformations from Lesson 2.

Add the cleaning stage below the read:

from pyspark.sql import functions as F

cleaned = (
    raw
    .filter(F.col("status") == "completed")
    .filter(F.col("amount").isNotNull())
    .filter(F.col("amount") > 0)
)
print("rows after cleaning:", cleaned.count())

Each filter is narrow — it drops rows without moving any data across the cluster. Together they remove the cancelled order (1004), the missing-amount order (1003), and the negative-amount order (1006).

Checkpoint: run it. You should see rows after cleaning: 9. The count is 9, not 7, because the two duplicate orders (1002 and 1005) are both still present — you remove those next. If you see 12, one of the filters is not being applied; check that all three .filter calls are chained onto raw.

Step 4 — Deduplicate to the latest row per order​

Goal: keep exactly one row per order_id — the one with the most recent event_time — using a window function.

Add the deduplication stage:

from pyspark.sql.window import Window

latest_first = Window.partitionBy("order_id").orderBy(
    F.col("event_time_ts").desc()
)

deduped = (
    cleaned
    .withColumn("event_time_ts", F.to_timestamp("event_time"))
    .withColumn("row_rank", F.row_number().over(latest_first))
    .filter(F.col("row_rank") == 1)
    .drop("row_rank", "event_time_ts")
)
print("rows after dedup:", deduped.count())
deduped.filter(F.col("order_id").isin("1002", "1005")).select(
    "order_id", "amount"
).show()

The window ranks each order's rows newest-first, and keeping row_rank == 1 selects the latest. For 1002 that keeps the 10:00 row with amount 320.0 over the 09:05 row; for 1005 it keeps the 09:20 row with 45.5 over the 08:00 row.

Checkpoint: run it. You should see rows after dedup: 7, then a table showing 1002 with amount 320.0 and 1005 with amount 45.5:

+--------+------+
|order_id|amount|
+--------+------+
|    1002| 320.0|
|    1005|  45.5|
+--------+------+

If 1002 shows 300.0, your window is ordering ascending — make sure the orderBy uses .desc().

Step 5 — Aggregate revenue per category​

Goal: collapse the clean orders into one row per category with total revenue and an order count — the wide step of the pipeline.

Add the aggregation:

summary = (
    deduped
    .groupBy("category")
    .agg(
        F.round(F.sum("amount"), 2).alias("revenue"),
        F.count("*").alias("orders"),
    )
    .orderBy("category")
)
summary.show()

This groupBy is the one wide transformation in the pipeline: Spark shuffles the seven clean rows so each category's rows meet, then sums within each group. Every step before it was narrow, so the shuffle moves only these few rows.

Checkpoint: run it. You should see exactly three category rows:

+-----------+-------+------+
|   category|revenue|orders|
+-----------+-------+------+
|      books|   60.0|     3|
|electronics|  820.0|     2|
|       home|  105.5|     2|
+-----------+-------+------+

If electronics shows 1120.0 with 3 orders, the duplicate 1002 was not removed — revisit the window in Step 4.

Step 6 — Write partitioned Parquet​

Goal: write the summary to Parquet, partitioned by category, so a reader can prune to one category without scanning the rest.

Add the write:

(
    summary.write
    .mode("overwrite")
    .partitionBy("category")
    .parquet("out/category_summary")
)
print("wrote out/category_summary")

partitionBy("category") lays the output out as one subfolder per category value. The category column is encoded in the folder name rather than stored in the files, and Spark restores it when the dataset is read back.

Checkpoint: run the script, then list the output folder.

python3 batch_pipeline.py
ls out/category_summary

You should see three partition folders — category=books, category=electronics, and category=home — alongside a _SUCCESS marker file. On Windows PowerShell, use Get-ChildItem out/category_summary instead of ls. If the folder is missing, confirm the write block ran without an error above it.

Validation​

Verify the whole build against the objective: correct partitioned Parquet, read back independently of the pipeline that wrote it. Save this as validate.py:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("de202-validate")
    .master("local[2]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

result = spark.read.parquet("out/category_summary")
rows = {
    r["category"]: (round(r["revenue"], 2), r["orders"])
    for r in result.collect()
}
expected = {
    "books": (60.0, 3),
    "electronics": (820.0, 2),
    "home": (105.5, 2),
}
total_revenue = round(sum(rev for rev, _ in rows.values()), 2)
total_orders = sum(cnt for _, cnt in rows.values())

if rows == expected and total_revenue == 985.5 and total_orders == 7:
    print(
        "PASSED — 3 categories, revenue 985.50 across 7 clean orders"
    )
else:
    print("FAILED — got", rows)

spark.stop()

Run it:

python3 validate.py

Checkpoint: the command prints PASSED — 3 categories, revenue 985.50 across 7 clean orders. If it prints FAILED, compare the printed dictionary against the expected three categories: books should total 60.0 over 3 orders, electronics 820.0 over 2, and home 105.5 over 2. Re-run python3 batch_pipeline.py first if you changed the pipeline after the last write.

Teardown​

Nothing here is billable — the whole lab runs on your machine. To reclaim space, remove the output and working folder with rm -rf out de202-batch-lab (macOS/Linux) or Remove-Item -Recurse out, de202-batch-lab (Windows). Keep the folder if you want batch_pipeline.py as a reference for the course's Spark Forge challenges. Deactivate the virtualenv with deactivate when you are done.

Troubleshooting​

SymptomLikely causeFix
Unable to locate a Java Runtime or JAVA_HOME is not setNo JDK installed or not on the pathInstall a JDK 11 or 17 and re-run; PySpark needs a JVM
ingested rows: 11 in Step 2A CSV data row is missingRe-save all twelve data rows from Step 1 exactly
rows after cleaning: 12 in Step 3A filter is not chained onto rawConfirm all three .filter calls are applied in one chain
1002 shows amount 300.0 in Step 4Window orderBy is ascendingOrder the window by event_time_ts with .desc()
electronics shows 1120.0 in Step 5Duplicate 1002 was not removedRe-check the row_number filter keeps row_rank == 1
Validation prints FAILEDStale output from an earlier runRe-run python3 batch_pipeline.py, then python3 validate.py