Batch ETL Pipeline: PySpark to PostgreSQL

Build a PySpark batch ETL pipeline that cleans Parquet data and loads it into PostgreSQL.

Introduction

30 Second Summary

Every online order generates a trail of raw data. In a real retail company, that raw data is full of duplicates, missing values, and cryptic product IDs that mean nothing to a business analyst.

In this project, you will build a batch ETL pipeline that transforms messy order data into clean, enriched analytics tables you can query for daily revenue by product category.

What You'll Build

You'll run a single command and watch PySpark read raw Parquet files, clean out duplicates and nulls, join orders with product names, and load daily revenue summaries into PostgreSQL. Every step is visible through live SQL queries that prove the transformation worked.

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

  • A working Docker Compose environment where you type one command and PySpark plus PostgreSQL spin up on a shared network, connected over JDBC so Spark writes directly to Postgres.
  • A data quality pipeline where you query PostgreSQL to see duplicates and nulls in raw data, run your cleaner, then query again to confirm they're gone.
  • A daily revenue KPI table where you query daily_kpis and see revenue broken down by product category and date, powered by a join between your orders and a product dimension table.
  • Secret Mission: Add 7-day rolling average revenue using PySpark window functions and load it as a trend analysis table.

Are there any prerequisites?

This project assumes you have completed Part 1 (Kafka Ingestion to Parquet) or are familiar with the Parquet file format. You should also have Docker Desktop installed and a code editor ready.

You do not need Java or PySpark installed on your machine. Both run inside Docker containers that this project builds for you.

Before We Start

Before We Start...

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

Set Up the Project Environment

Your batch ETL pipeline needs two services working together: PySpark to transform data and PostgreSQL to store the results.

Docker Compose lets you define both services in one configuration file so they run on a shared network. This means Spark can write directly to Postgres via JDBC without you installing Java or Postgres on your machine.

In this step, get ready to:
  • Create Docker configuration files that define your PySpark and PostgreSQL environment.
  • Create sample data files including a product dimension table and a data generator script.
  • Build the containers and generate sample Parquet data with intentional quality issues.
Create the project directory and Docker configuration

You need a dedicated folder on your Desktop to hold all your project files.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Terminal, and pressing Enter.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir batch-etl-pipeline
cd batch-etl-pipeline
  • Confirm the folder was created by running:
pwd

You should see a path ending in batch-etl-pipeline.

The first file to create is docker-compose.yml, which defines both services and their shared network.

  • In your code editor, create a new file called docker-compose.yml inside the batch-etl-pipeline folder.
  • Paste the PostgreSQL service definition into docker-compose.yml:
services:
  postgres:
    image: postgres:16
    container_name: retail-postgres
    environment:
      POSTGRES_USER: spark
      POSTGRES_PASSWORD: spark123
      POSTGRES_DB: retail
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U spark -d retail"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - etl-network

What does this configure?

  • image: postgres:16 pulls the official PostgreSQL 16 image from Docker Hub.
  • The environment block creates a database called retail with credentials the Spark container will use to connect.
  • The healthcheck runs pg_isready every 5 seconds so Spark only starts after Postgres is accepting connections.
  • Below the postgres service (still inside docker-compose.yml), add the Spark service, network, and volume definitions:

  spark:
    build: .
    container_name: retail-spark
    volumes:
      - ./:/app
    depends_on:
      postgres:
        condition: service_healthy
    networks:
      - etl-network

networks:
  etl-network:
    driver: bridge

volumes:
  postgres_data:

What does the Spark service do?

build: . tells Docker to build the Spark image from the Dockerfile in the same folder.

volumes: ./:/app mounts your entire project folder into the container at /app. This means any file you create locally is instantly available inside the container.

The etl-network bridge lets the Spark container reach Postgres by its service name (postgres) as a hostname.

depends_on: service_healthy ensures Spark waits for the Postgres healthcheck to pass before starting.

  • Save docker-compose.yml.
  • Validate that Docker can parse your Compose file by running:
docker compose config --services

You should see two service names printed:

postgres
spark

Seeing a YAML parsing error?

  • Check that indentation uses spaces (not tabs). YAML requires consistent 2-space indentation.
  • Make sure the spark service sits at the same indent level as postgres (both under services:).
  • Verify that networks: and volumes: are at the root level (no indentation), not nested under services:.

Help me fix a YAML error in my docker-compose.yml

Now create the Dockerfile that builds the Spark container image.

  • Create a new file called Dockerfile (no extension) in the batch-etl-pipeline folder and paste this content:
FROM python:3.11-slim

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

RUN pip install --no-cache-dir pyspark==3.5.9 pyarrow pandas

RUN mkdir -p /opt/spark/jars && \
    wget -q -P /opt/spark/jars/ \
    https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.13/postgresql-42.7.13.jar

WORKDIR /app

CMD ["tail", "-f", "/dev/null"]

What does this Dockerfile install?

  • python:3.11-slim gives you a lightweight Python base image.
  • openjdk-17-jre-headless installs Java 17, which PySpark needs to run its Spark engine under the hood.
  • pyspark==3.5.9 is the PySpark library pinned to a stable version.
  • The postgresql-42.7.13.jar is the JDBC driver that lets Spark talk directly to PostgreSQL.
  • tail -f /dev/null keeps the container running so you can exec commands into it.
  • Create one more file called init.sql in the same folder and paste this content:
-- Schema is created automatically by Spark's JDBC writer
-- This file ensures the database is ready
SELECT 'retail database ready' AS status;
  • Save both Dockerfile and init.sql.
  • Confirm all configuration files exist by running:
ls

You should see Dockerfile, docker-compose.yml, and init.sql listed.

Missing a file?

Make sure you saved each file inside the batch-etl-pipeline folder (not one level up on the Desktop). If a file is missing, check your editor's file explorer to confirm the save location.

Help me figure out why my project files aren't showing up in the right folder

✔️ Awesome, I've got everything!

Great. Double check you've saved all three files.

ⓧ I'd like to double check the full code

Your docker-compose.yml should contain the postgres service (chunk 1) followed by the spark service, networks, and volumes (chunk 2). Scroll up and confirm both sections are present in order.

Your Dockerfile should be:

FROM python:3.11-slim

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

RUN pip install --no-cache-dir pyspark==3.5.9 pyarrow pandas

RUN mkdir -p /opt/spark/jars && \
    wget -q -P /opt/spark/jars/ \
    https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.13/postgresql-42.7.13.jar

WORKDIR /app

CMD ["tail", "-f", "/dev/null"]

Your init.sql should be:

-- Schema is created automatically by Spark's JDBC writer
-- This file ensures the database is ready
SELECT 'retail database ready' AS status;
Create the data files

Your pipeline needs two data inputs: a product lookup table and a script that generates realistic order data with intentional quality problems.

  • Create a new file called products.csv in the batch-etl-pipeline folder and paste this content:
product_id,product_name,category,unit_cost
1,Wireless Mouse,Electronics,12.99
2,USB-C Cable,Electronics,8.49
3,Mechanical Keyboard,Electronics,54.99
4,Laptop Stand,Electronics,29.99
5,Running Shoes,Sportswear,89.99
6,Yoga Mat,Sportswear,24.99
7,Water Bottle,Sportswear,14.99
8,Dumbbell Set,Sportswear,49.99
9,Coffee Beans 1kg,Grocery,18.99
10,Olive Oil 500ml,Grocery,9.99
11,Protein Bars 12pk,Grocery,22.99
12,Green Tea Box,Grocery,6.99
13,Desk Lamp,Home,34.99
14,Throw Pillow,Home,19.99
15,Scented Candle,Home,12.99
16,Wall Clock,Home,27.99

What is a dimension table?

A dimension table maps IDs to human-readable attributes. Raw order events only store a product_id number. By joining with this table later, you can group revenue by category names like Electronics or Sportswear instead of meaningless IDs.

  • Save products.csv.
  • Confirm it exists by running:
ls

You should now see products.csv alongside your Docker files.

Next, create the data generation script that produces messy Parquet files simulating real-world pipeline output.

  • Create a new file called generate_data.py in the batch-etl-pipeline folder and paste this first section (imports and base order generation):
import pandas as pd
import numpy as np
import os
from datetime import datetime, timedelta

# Set random seed for reproducible data
np.random.seed(42)

# Configuration
num_orders = 1000
num_products = 16
num_customers = 200

# Generate random order dates across July 2026
start_date = datetime(2026, 7, 1)
dates = [start_date + timedelta(days=int(d)) for d in np.random.randint(0, 30, num_orders)]

# Build the base orders DataFrame
orders = pd.DataFrame({
    "order_id": range(1, num_orders + 1),
    "customer_id": np.random.randint(1, num_customers + 1, num_orders),
    "product_id": np.random.randint(1, num_products + 1, num_orders),
    "quantity": np.random.randint(1, 6, num_orders),
    "price": np.round(np.random.uniform(5.0, 100.0, num_orders), 2),
    "order_timestamp": [d.strftime("%Y-%m-%d %H:%M:%S") for d in dates],
})

What does this code do?

This creates a DataFrame of 1000 fake retail orders with random customer IDs, product IDs, quantities, and prices. The np.random.seed(42) ensures everyone gets the same "random" data, making debugging easier.

  • Below the DataFrame creation in generate_data.py, add the data quality injection and Parquet writing logic:
# Inject data quality issues: ~50 duplicate rows
duplicate_indices = np.random.choice(num_orders, size=50, replace=False)
duplicates = orders.iloc[duplicate_indices].copy()
orders = pd.concat([orders, duplicates], ignore_index=True)

# Inject ~30 null product_ids
null_indices = np.random.choice(len(orders), size=30, replace=False)
orders.loc[null_indices, "product_id"] = None

# Shuffle rows
orders = orders.sample(frac=1, random_state=42).reset_index(drop=True)

# Write to Parquet
output_dir = "/app/data/raw_orders"
os.makedirs(output_dir, exist_ok=True)

# Split into 3 Parquet files (simulating batched consumer output)
chunk_size = len(orders) // 3
for i in range(3):
    start = i * chunk_size
    end = start + chunk_size if i < 2 else len(orders)
    chunk = orders.iloc[start:end]
    chunk.to_parquet(
        f"{output_dir}/orders_batch_{i:03d}.parquet",
        index=False,
        engine="pyarrow",
    )

print(f"Generated {len(orders)} order records across 3 Parquet files")
print(f"  - Includes ~50 duplicate order_ids")
print(f"  - Includes ~30 null product_ids")
print(f"  - Output: {output_dir}/")

Why inject bad data on purpose?

Real-world data pipelines always have quality problems. Upstream systems produce duplicates from retries. Missing values sneak in from incomplete API responses.

By injecting ~50 duplicate orders and ~30 null product IDs, you create a realistic dataset that you will discover and fix in Step 3.

  • Save generate_data.py.
  • Confirm all project files are in place by running:
ls

You should see: Dockerfile, docker-compose.yml, generate_data.py, init.sql, and products.csv.

✔️ Awesome, I've got everything!

Great. Double check you've saved generate_data.py with both sections (imports + DataFrame, then injection + Parquet write).

ⓧ I'd like to double check the full code

Your generate_data.py should contain both chunks in order: the imports and DataFrame creation (chunk 1 above) followed by the injection and Parquet writing (chunk 2 above). Scroll up and confirm both sections are present, in order.

Your products.csv should be:

product_id,product_name,category,unit_cost
1,Wireless Mouse,Electronics,12.99
2,USB-C Cable,Electronics,8.49
3,Mechanical Keyboard,Electronics,54.99
4,Laptop Stand,Electronics,29.99
5,Running Shoes,Sportswear,89.99
6,Yoga Mat,Sportswear,24.99
7,Water Bottle,Sportswear,14.99
8,Dumbbell Set,Sportswear,49.99
9,Coffee Beans 1kg,Grocery,18.99
10,Olive Oil 500ml,Grocery,9.99
11,Protein Bars 12pk,Grocery,22.99
12,Green Tea Box,Grocery,6.99
13,Desk Lamp,Home,34.99
14,Throw Pillow,Home,19.99
15,Scented Candle,Home,12.99
16,Wall Clock,Home,27.99
Build the containers and generate sample data

All your configuration is in place. Time to build the Docker image and start both containers.

  • Build and start both containers in detached mode by running:
docker compose up -d --build

What does this command do?

  • --build forces Docker to build the Spark image from your Dockerfile (installing Java, PySpark, and the JDBC driver).
  • -d runs everything in the background so your terminal stays free.
  • Docker will pull the postgres:16 image, build the Spark image, then start both on the etl-network.

First build takes a few minutes

Docker needs to download the Python base image, install Java, and download the JDBC jar. Subsequent builds will be much faster because Docker caches each layer.

  • Verify both containers are running by checking their status:
docker compose ps

You should see two containers (retail-postgres and retail-spark) with a status of Up. The postgres container should also show (healthy).

Container not healthy or exited?

  • If postgres shows (health: starting), wait 15 seconds and run docker compose ps again. The healthcheck runs every 5 seconds.
  • If a container exited, check its logs with docker compose logs postgres or docker compose logs spark.
  • Make sure Docker Desktop is running. If it crashed, restart it and run docker compose up -d --build again.

Help me troubleshoot my Docker containers that aren't starting

Now generate the sample Parquet data inside the Spark container.

  • Run the data generation script inside the Spark container:
docker compose exec spark python generate_data.py

You should see output confirming the data was created:

Generated 1050 order records across 3 Parquet files
  - Includes ~50 duplicate order_ids
  - Includes ~30 null product_ids
  - Output: /app/data/raw_orders/

Why run inside the container?

The script writes files to /app/data/raw_orders/ inside the container. Because your project folder is volume-mounted, these files also appear locally in batch-etl-pipeline/data/raw_orders/. Running inside the container ensures the paths match what PySpark will use later.

Before you run the final verification commands, what do you think will happen if you try to import PySpark inside the container?

  • Verify PySpark is working inside the Spark container by running:
docker compose exec spark python -c "from pyspark.sql import SparkSession; print('PySpark ready')"

You should see PySpark ready printed.

  • Verify PostgreSQL is accepting connections by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT 1;"

You should see a table with a single column and the value 1.

PySpark import failing?

  • If you see ModuleNotFoundError, the Docker image may not have built correctly. Run docker compose down then docker compose up -d --build to rebuild from scratch.
  • If the exec command says the container is not running, check docker compose ps and restart with docker compose up -d.

Help me debug PySpark not being found inside my Docker container

Your environment is ready. Both containers are running, PySpark can import successfully, PostgreSQL is healthy, and you have messy Parquet files waiting to be processed. Next up, you'll load that raw data into Spark and push it straight to PostgreSQL.

Load Raw Parquet and Write to PostgreSQL

Your containers are running and your sample data is generated. Now you need to actually see that data.

Before you can fix data quality issues like duplicates and nulls, you need to load the raw Parquet files and inspect what's inside them. In this step, you'll write a PySpark script that reads Parquet into a DataFrame, pushes the raw data straight to PostgreSQL, and proves the JDBC connection works end-to-end.

In this step, get ready to:
  • Create a PySpark script with a SparkSession configured for the JDBC driver.
  • Read raw Parquet files into a DataFrame and inspect the schema.
  • Write the raw DataFrame to PostgreSQL and verify the row count.
Create the batch ETL script and load Parquet data

Every PySpark job starts with a SparkSession. This is the entry point that configures how Spark runs, where to find the JDBC driver jar, and how much memory to allocate.

  • Create a new file called batch_etl.py in your project folder (the same folder as docker-compose.yml).
  • Add the following code to set up the Spark session and JDBC connection properties:
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count, to_date

# Initialize Spark session
spark = SparkSession.builder \
    .appName("RetailBatchETL") \
    .master("local[*]") \
    .config("spark.jars", "/opt/spark/jars/postgresql-42.7.13.jar") \
    .config("spark.driver.memory", "1g") \
    .getOrCreate()

# JDBC connection properties
jdbc_url = "jdbc:postgresql://postgres:5432/retail"
jdbc_properties = {
    "user": "spark",
    "password": "spark123",
    "driver": "org.postgresql.Driver",
}

What does this code do?

  • SparkSession.builder creates the entry point for all PySpark operations. local[*] tells Spark to use all available CPU cores inside the container.
  • spark.jars points to the PostgreSQL JDBC driver jar that was pre-downloaded in the Dockerfile. Without this, Spark cannot talk to Postgres.
  • jdbc_url uses the container name postgres as the hostname because both containers share the etl-network bridge network.
  • jdbc_properties holds the login credentials and the JDBC driver class name that Spark needs to establish the connection.
  • Save batch_etl.py.

Now add the code that reads all three Parquet files and displays what's inside them.

  • Add the following code below the jdbc_properties dictionary in batch_etl.py:
# --- Step 2: Load raw Parquet and write to PostgreSQL ---
print("=== Loading raw Parquet files ===")
raw_df = spark.read.parquet("/app/data/raw_orders/")
raw_df.printSchema()
raw_df.show(10)
print(f"Raw record count: {raw_df.count()}")

What does this code do?

  • spark.read.parquet("/app/data/raw_orders/") reads ALL Parquet files in that directory into a single DataFrame. Spark automatically discovers and combines the three batch files.
  • printSchema() prints a tree showing each column's name, data type, and whether it allows nulls.
  • show(10) displays the first 10 rows as a formatted table so you can eyeball the data.
  • count() triggers Spark to scan all rows and return the total. This number will include the intentional duplicates and nulls.
  • Save batch_etl.py.
  • Run the script inside the Spark container by running this command:
docker compose exec spark python batch_etl.py

You should see a schema tree printed followed by 10 rows of order data and a raw record count of approximately 1080.

Seeing a Java or class not found error?

  • Make sure the spark.jars path matches the jar location in the Dockerfile: /opt/spark/jars/postgresql-42.7.13.jar. A typo in the path causes Spark to silently skip the driver.
  • If you see No such file or directory for the Parquet path, confirm that generate_data.py ran successfully in Step 1. Re-run it with docker compose exec spark python generate_data.py.

Help me debug my PySpark script not finding the JDBC jar or Parquet files.

Write raw data to PostgreSQL via JDBC

The DataFrame is loaded in memory, but it's not queryable from outside Spark yet. Writing it to PostgreSQL via JDBC makes the data accessible to any SQL tool.

  • Add the following code at the bottom of batch_etl.py, below the print(f"Raw record count: {raw_df.count()}") line:
raw_df.write.jdbc(
    url=jdbc_url,
    table="raw_orders",
    mode="overwrite",
    properties=jdbc_properties,
)
print("Raw orders written to PostgreSQL: raw_orders")

What does this code do?

  • raw_df.write.jdbc() pushes the entire DataFrame to PostgreSQL in one call. Spark handles creating the table and inserting all rows.
  • mode="overwrite" drops the table if it already exists and recreates it. This is safe for development because you can re-run the script without conflicts.
  • The jdbc_properties dictionary passes the username, password, and driver class that PostgreSQL needs to authenticate the connection.
  • Save batch_etl.py.
Run the pipeline and verify the row count

Time to run the complete script and confirm data actually landed in PostgreSQL.

  • Run the full ETL script inside the Spark container:
docker compose exec spark python batch_etl.py

You should see the schema tree, 10 sample rows, the raw record count, and a final message confirming the write to PostgreSQL.

Before you query PostgreSQL, predict: how many rows do you expect to find in the raw_orders table? Remember, the data generator created 1000 base orders plus approximately 50 duplicates.

  • Verify the row count in PostgreSQL by running this command:
docker compose exec postgres psql -U spark -d retail -c "SELECT count(*) FROM raw_orders;"

You should see approximately 1050 rows. The exact count depends on how the random duplicates and null-injection affected the final dataset, but it will be more than 1000.

Seeing zero rows or a missing table error?

  • Check that batch_etl.py printed the confirmation message Raw orders written to PostgreSQL: raw_orders when you ran it. If it's missing, the JDBC write failed silently.
  • Confirm the JDBC URL uses the container name postgres and port 5432. Using localhost instead will fail because the Spark container cannot reach Postgres via localhost.
  • Make sure both containers are on the same network by running docker network inspect etl-network.

Help me figure out why my JDBC write to PostgreSQL failed.

✔️ Awesome, I've got everything!

Great. Double check you've saved batch_etl.py and your raw_orders table has approximately 1050 rows.

ⓧ I'd like to double check the full code

Here's the complete batch_etl.py at this point:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count, to_date

# Initialize Spark session
spark = SparkSession.builder \
    .appName("RetailBatchETL") \
    .master("local[*]") \
    .config("spark.jars", "/opt/spark/jars/postgresql-42.7.13.jar") \
    .config("spark.driver.memory", "1g") \
    .getOrCreate()

# JDBC connection properties
jdbc_url = "jdbc:postgresql://postgres:5432/retail"
jdbc_properties = {
    "user": "spark",
    "password": "spark123",
    "driver": "org.postgresql.Driver",
}

# --- Step 2: Load raw Parquet and write to PostgreSQL ---
print("=== Loading raw Parquet files ===")
raw_df = spark.read.parquet("/app/data/raw_orders/")
raw_df.printSchema()
raw_df.show(10)
print(f"Raw record count: {raw_df.count()}")

raw_df.write.jdbc(
    url=jdbc_url,
    table="raw_orders",
    mode="overwrite",
    properties=jdbc_properties,
)
print("Raw orders written to PostgreSQL: raw_orders")

Your end-to-end pipeline is working. Raw Parquet data flows from disk through PySpark into PostgreSQL. Next up, you'll discover the data quality problems hiding in those rows and fix them.

Discover and Fix Data Quality Issues

In the last step, you loaded raw Parquet data into PostgreSQL and proved the end-to-end JDBC pipeline works. You now have a table with over a thousand rows ready for analysis.

There is a problem hiding in that data. The raw_orders table contains duplicate rows (the same order appearing multiple times) and rows with missing product IDs. If you ran analytics on this messy data, your revenue numbers would be inflated and your joins would silently drop rows.

In this step, get ready to:
  • Query PostgreSQL to discover duplicates and null values in the raw data.
  • Add PySpark cleaning logic to deduplicate and filter out bad rows.
  • Verify the cleaned data has exactly 1000 valid rows in PostgreSQL.
Query PostgreSQL to expose data quality problems

Before writing any fix, you need to see the problems with your own eyes. You will query the raw_orders table directly in PostgreSQL to find out how many duplicates and null values exist.

  • Check for duplicate order IDs by running this query:
docker compose exec postgres psql -U spark -d retail -c "SELECT order_id, count(*) FROM raw_orders GROUP BY order_id HAVING count(*) > 1 LIMIT 10;"

You should see a table showing several order_id values that each appear 2 or more times. These are the duplicate rows that would inflate your analytics.

  • Count exactly how many rows have a null product ID by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT count(*) FROM raw_orders WHERE product_id IS NULL;"

You should see a count of approximately 30. These rows have no product ID, meaning they cannot be joined to a product lookup table later.

Why do duplicates and nulls matter?

Duplicate orders mean a single purchase gets counted (and summed) multiple times in revenue calculations. A report built on this data would overstate revenue by roughly 5%.

Null product IDs break any downstream join with a product dimension table. Those rows silently vanish from results, making your totals inconsistent.

Add cleaning logic to the ETL script

Now that you have seen the problems, you will fix them with PySpark DataFrame operations. The cleaning pipeline will remove duplicates, filter nulls, and cast columns to their proper types.

  • In batch_etl.py, find the last line that says:
print("Raw orders written to PostgreSQL: raw_orders")
  • Add the following cleaning code below that line:
# --- Step 3: Clean and deduplicate ---
print("\n=== Cleaning data ===")
# Remove duplicates based on order_id
deduped_df = raw_df.dropDuplicates(["order_id"])
print(f"After deduplication: {deduped_df.count()} records")

# Remove rows with null product_id
clean_df = deduped_df.filter(col("product_id").isNotNull())
print(f"After null removal: {clean_df.count()} records")

# Cast types for consistency
clean_df = clean_df.withColumn("product_id", col("product_id").cast("integer")) \
                   .withColumn("order_timestamp", col("order_timestamp").cast("timestamp"))

clean_df.write.jdbc(
    url=jdbc_url,
    table="clean_orders",
    mode="overwrite",
    properties=jdbc_properties,
)
print("Clean orders written to PostgreSQL: clean_orders")

What does this code do?

  • dropDuplicates(["order_id"]) keeps only the first occurrence of each order ID, dropping any repeated rows.
  • filter(col("product_id").isNotNull()) removes any row where the product ID is missing, since those rows cannot be enriched with product data later.
  • The withColumn calls cast product_id to integer and order_timestamp to a proper timestamp type, ensuring consistent types for downstream joins and aggregations.
  • Finally, the cleaned DataFrame is written to a new clean_orders table in PostgreSQL via JDBC.
  • Save batch_etl.py.

Not sure where to add the code?

The new cleaning section goes immediately after the raw orders write block. It should be the last code in the file at this point. Make sure raw_df is still in scope because the cleaning code references it directly.

help me figure out where to add the cleaning code in batch_etl.py

Run the ETL and verify clean data

Time to see if your cleaning logic actually works. You will re-run the full ETL script, then query PostgreSQL to confirm the problems are gone.

  • Run the updated ETL script by executing:
docker compose exec spark python batch_etl.py

In the terminal output, look for the cleaning section. You should see "After deduplication: 1000 records" and "After null removal: 1000 records" (since the ~30 nulls were all on duplicate rows that were already removed).

Before you verify in PostgreSQL, what row count do you expect in clean_orders?

  • Verify the row count in the clean_orders table by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT count(*) FROM clean_orders;"

You should see exactly 1000. That confirms the duplicates are gone and the original 1000 unique orders are preserved.

  • Confirm there are zero duplicates in the clean table by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT order_id, count(*) FROM clean_orders GROUP BY order_id HAVING count(*) > 1;"

You should see zero rows returned. Every order ID now appears exactly once.

  • Confirm there are zero null product IDs by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT count(*) FROM clean_orders WHERE product_id IS NULL;"

You should see a count of 0. Every row in clean_orders has a valid product ID, ready for enrichment.

Seeing more than 1000 rows?

Make sure you used dropDuplicates(["order_id"]) with the column name in a list. Calling dropDuplicates() with no arguments deduplicates on ALL columns, which only catches rows that are identical in every field.

Also check that the filter step uses isNotNull() and not isNull(). Using isNull() would keep only the null rows instead of removing them.

help me debug why my clean_orders table has the wrong row count

✔️ Awesome, I've got everything!

Great. Make sure you have saved batch_etl.py and that your clean_orders table shows 1000 rows with zero duplicates and zero nulls.

ⓧ I'd like to double check the full code

Your batch_etl.py should now have two main sections in order:

  • The Step 2 block: SparkSession setup, JDBC config, raw Parquet load, and write to raw_orders.
  • The Step 3 block you just added: the deduplication with dropDuplicates(["order_id"]), null filter with filter(col("product_id").isNotNull()), type casting, and write to clean_orders.

Scroll up to confirm both code blocks are present in your file, in that order.

Your data is now clean and trustworthy. Next up, you will enrich these orders with product names and categories, then compute daily revenue KPIs that a business analyst could actually use.

Enrich with Product Dimension and Aggregate KPIs

Your clean_orders table now has 1000 rows of deduplicated, null-free order data. That is a solid foundation, but an order with product_id = 7 means nothing to a business analyst staring at a dashboard.

In this step, you will join your clean orders with a product dimension table to add human-readable names and categories. Then you will aggregate the enriched data into daily revenue KPIs that could power a real analytics dashboard.

In this step, get ready to:
  • Join clean orders with product metadata to create an enriched orders table.
  • Compute daily revenue KPIs grouped by product category.
  • Verify all four PostgreSQL tables with correct row counts and meaningful data.
Join clean orders with the product dimension

A dimension table holds descriptive attributes about an entity. Your products.csv file maps each product_id to a name, category, and unit cost.

By joining this lookup table onto your clean orders, every row gains context that makes it queryable by business users.

  • In batch_etl.py, find the line that says print("Clean orders written to PostgreSQL: clean_orders").
  • Add the following code below that line to load the product dimension and join it with your clean orders:
# --- Step 4: Enrich with product dimension ---
print("\n=== Enriching with product dimension ===")
products_df = spark.read.csv("/app/products.csv", header=True, inferSchema=True)
products_df.show(5)

enriched_df = clean_df.join(products_df, on="product_id", how="inner")
print(f"Enriched record count: {enriched_df.count()}")

enriched_df.write.jdbc(
    url=jdbc_url,
    table="enriched_orders",
    mode="overwrite",
    properties=jdbc_properties,
)
print("Enriched orders written to PostgreSQL: enriched_orders")

What does this code do?

  • spark.read.csv("/app/products.csv", header=True, inferSchema=True) loads the product lookup table into a PySpark DataFrame, automatically detecting column types from the data.
  • clean_df.join(products_df, on="product_id", how="inner") matches every order to its product details. An inner join drops any orders whose product_id does not exist in the lookup table.
  • The enriched DataFrame now includes product_name, category, and unit_cost columns alongside the original order columns.
  • Save batch_etl.py.
  • Run the ETL script to test the enrichment by running this command:
docker compose exec spark python batch_etl.py

In the output, look for Enriched record count: 1000 and a preview of your products table showing five rows with product_id, product_name, category, and unit_cost columns.

  • Verify the enriched data landed in PostgreSQL by running this query:
docker compose exec postgres psql -U spark -d retail -c "SELECT product_name, category, quantity, price FROM enriched_orders LIMIT 5;"

You should see five rows with real product names like Wireless Mouse and categories like Electronics attached to each order.

Seeing zero rows or an error?

  • If you see relation "enriched_orders" does not exist, make sure you saved batch_etl.py before running it. Re-run the script and try the query again.
  • If the enriched count is lower than 1000, check that you placed the enrichment code AFTER the cleaning section. The join uses clean_df which is only available after the deduplication logic.

Still stuck? Ask for help.

Compute daily revenue KPIs per category

Enriched orders tell you what happened on each individual transaction. But a dashboard needs aggregated metrics: how much revenue did each category generate per day?

  • In batch_etl.py, add the following code below the line print("Enriched orders written to PostgreSQL: enriched_orders") to compute and write daily KPIs:
# --- Step 4: Compute daily KPIs ---
print("\n=== Computing daily KPIs ===")
daily_kpis = enriched_df \
    .withColumn("order_date", to_date(col("order_timestamp"))) \
    .groupBy("order_date", "category") \
    .agg(
        count("order_id").alias("order_count"),
        spark_sum(col("quantity") * col("price")).alias("total_revenue"),
    ) \
    .orderBy("order_date", "category")

daily_kpis.show(10)

daily_kpis.write.jdbc(
    url=jdbc_url,
    table="daily_kpis",
    mode="overwrite",
    properties=jdbc_properties,
)
print("Daily KPIs written to PostgreSQL: daily_kpis")

print("\n=== ETL pipeline complete ===")
print("Tables created: raw_orders, clean_orders, enriched_orders, daily_kpis")

spark.stop()

What does this code do?

  • to_date(col("order_timestamp")) extracts just the date portion from each timestamp, so orders on the same day group together regardless of time.
  • groupBy("order_date", "category") creates one group for every unique combination of date and product category.
  • The agg() call computes two metrics per group: order_count (number of orders) and total_revenue (sum of quantity multiplied by price).
  • spark.stop() shuts down the Spark session cleanly, releasing memory inside the container.

Why use spark_sum instead of sum?

Python has a built-in sum function. The import line at the top of the file renames PySpark's sum to spark_sum to avoid a name collision. This is a common pattern in PySpark projects.

  • Save batch_etl.py.
  • Run the updated pipeline by executing this command:
docker compose exec spark python batch_etl.py

In the output, you should see a preview table with order_date, category, order_count, and total_revenue columns, followed by the message ETL pipeline complete.

Seeing an AnalysisException or missing column error?

  • If you see an error about order_timestamp not being found, make sure the KPI code uses enriched_df (which inherits columns from clean_df), not products_df.
  • Double check that your KPI code block comes AFTER the enrichment block. The variable enriched_df must be defined before you use it.

Need help debugging this error?

Verify all four tables end-to-end

Your complete ETL pipeline now runs four stages: raw load, cleaning, enrichment, and aggregation. Time to verify the final state of all tables in PostgreSQL.

  • Check the row counts for all four tables by running this command:
docker compose exec postgres psql -U spark -d retail -c "SELECT 'raw_orders' AS tbl, count(*) FROM raw_orders UNION ALL SELECT 'clean_orders', count(*) FROM clean_orders UNION ALL SELECT 'enriched_orders', count(*) FROM enriched_orders UNION ALL SELECT 'daily_kpis', count(*) FROM daily_kpis;"

You should see approximately 1080 raw_orders, 1000 clean_orders, 1000 enriched_orders, and roughly 120 daily_kpis rows (30 days multiplied by 4 categories).

Before you run the final query below, predict: what columns will the daily_kpis table contain, and will the revenue numbers look realistic for a retail store?

  • Query the top daily revenue entries by running:
docker compose exec postgres psql -U spark -d retail -c "SELECT * FROM daily_kpis ORDER BY order_date DESC, total_revenue DESC LIMIT 5;"

You should see rows with order_date, category, order_count, and total_revenue columns populated with realistic values. Categories like Electronics and Sportswear appear alongside revenue totals in the hundreds.

Why is this the final shape that matters?

The daily_kpis table is what a BI tool like Metabase or Grafana would query to render a revenue chart. One SQL query against this pre-aggregated table replaces a complex join and group that would otherwise run on every dashboard refresh.

✔️ Awesome, I've got everything!

Great. Double check you have saved batch_etl.py and that all four tables exist in PostgreSQL.

ⓧ I'd like to double check the full code

Your batch_etl.py should contain these sections in order. Scroll up and confirm each one is present:

  • Imports (SparkSession, col, spark_sum, count, to_date).
  • SparkSession builder with RetailBatchETL app name and JDBC jar config.
  • JDBC connection properties (jdbc_url, jdbc_properties).
  • Step 2: Load raw Parquet and write to raw_orders.
  • Step 3: Deduplicate, filter nulls, cast types, write to clean_orders.
  • Step 4: Load products.csv, inner join with clean_df, write to enriched_orders.
  • Step 4: Group by order_date and category, aggregate KPIs, write to daily_kpis.
  • Pipeline complete message and spark.stop().

Your pipeline now transforms messy raw events into clean, enriched, and aggregated analytics tables. Next up, you will push this further with rolling window calculations in the Secret Mission.

Secret mission

Add Rolling Window Analytics

Static daily totals are useful, but they don't reveal trends. A 7-day rolling average smooths out daily revenue spikes and gives analysts a trend line. Can you extend your pipeline with PySpark Window functions to compute rolling averages and load them into a new PostgreSQL table?

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 using Docker, so there are no ongoing cloud costs.

Resources you used:

  • Docker containers: retail-spark and retail-postgres running on the etl-network bridge network.
  • Docker volume: postgres_data storing PostgreSQL data (persists across container restarts).
  • Local project folder with source files (docker-compose.yml, Dockerfile, batch_etl.py, generate_data.py, products.csv, init.sql) and generated Parquet data.

Keep everything running

No action needed. Choose this if you're still actively building on the pipeline or plan to extend it soon.

  • Your containers stay running and PostgreSQL retains all five tables (raw_orders, clean_orders, enriched_orders, daily_kpis, rolling_kpis).
  • Re-run the ETL anytime with docker compose exec spark python batch_etl.py.
  • Docker Desktop uses some memory while containers are running, but there are no financial costs.

Pause - I'll come back to this later

Shut down running containers to free up memory. Your PostgreSQL data and project files are preserved.

  • Stop both containers by running this command from your project folder:
docker compose stop
  • When you're ready to resume, start them again with:
docker compose start
  • Your PostgreSQL data persists in the postgres_data volume, so all five tables will still be there when you resume.

Delete - I don't want to use this again

Remove all Docker containers, the PostgreSQL volume, and your local project files entirely.

  • From your project folder, remove containers, networks, and the PostgreSQL volume by running:
docker compose down -v
  • This removes both containers (retail-spark and retail-postgres), the etl-network, and the postgres_data volume with all database contents.
  • Delete the project folder from your Desktop by running:

macOS

cd ~/Desktop
rm -rf batch-etl-pipeline

Windows

cd ~\Desktop
rmdir /s /q batch-etl-pipeline
  • Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see the project folder listed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a complete batch ETL pipeline that transforms messy raw data into clean, queryable analytics tables in PostgreSQL.

Here's what you accomplished:

  • Built a containerized PySpark batch ETL pipeline that reads Parquet files, discovers data quality issues (duplicates and nulls), and fixes them with DataFrame operations like dropDuplicates and filter.
  • Joined fact data with a product dimension table and computed daily revenue KPIs per category using PySpark's groupBy and agg operations.
  • Wrote PySpark DataFrames directly to PostgreSQL via JDBC, verifying results with psql queries to confirm row counts and data integrity.
  • Secret Mission: Implemented 7-day rolling average revenue using PySpark Window functions for trend analysis.

Ready to quiz yourself?