Spark ETL Pipeline Project

ETL means extract, transform, and load. A pipeline pulls raw data from a source, cleans and reshapes it, and stores the result where analysts can use it. The project below ties together reading, schemas, cleaning, joins, aggregation, quality checks, and partitioned output in one realistic job.

The Layman View

A juice factory receives fruit crates, washes and sorts the fruit, presses the juice, and bottles it. Each station depends on the one before. A data pipeline treats raw files like fruit crates and ends with clean, labelled tables.

The Business Goal

An online shop wants a daily report of revenue per country and product category. Two raw files arrive every day.

orders.csv                              customers.csv
order_id, customer_id, product,         customer_id, name, country
category, quantity, price, order_date

Pipeline Overview

EXTRACT                 TRANSFORM                          LOAD
-------                 ---------                          ----
orders.csv    ---\      clean --> join --> aggregate  --->  Parquet report
customers.csv ---->                |                        partitioned by date
                              quality checks

Step 1: Extract with Explicit Schemas

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("daily_revenue").getOrCreate()

orders_schema = ("order_id STRING, customer_id STRING, product STRING, "
                 "category STRING, quantity INT, price DECIMAL(10,2), order_date DATE")
customers_schema = "customer_id STRING, name STRING, country STRING"

orders = spark.read.csv("/data/raw/orders.csv", header=True, schema=orders_schema)
customers = spark.read.csv("/data/raw/customers.csv", header=True, schema=customers_schema)

Step 2: Clean the Orders

Raw data holds duplicates, missing values, and impossible numbers. Remove them early so later steps run on trustworthy rows.

clean_orders = (orders
    .dropDuplicates(["order_id"])
    .filter(F.col("order_id").isNotNull() & F.col("customer_id").isNotNull())
    .filter((F.col("quantity") > 0) & (F.col("price") > 0))
    .withColumn("category", F.initcap(F.trim("category")))
    .withColumn("revenue", F.col("quantity") * F.col("price")))

Step 3: Join Customer Details

The customer table is small, so a broadcast join avoids a large shuffle.

enriched = clean_orders.join(F.broadcast(customers), "customer_id", "left") \
                       .withColumn("country", F.coalesce("country", F.lit("Unknown")))

Step 4: Aggregate

report = (enriched
    .groupBy("order_date", "country", "category")
    .agg(F.sum("revenue").alias("total_revenue"),
         F.countDistinct("order_id").alias("orders"),
         F.sum("quantity").alias("units")))

Rows in clean_orders: one per order
        |
groupBy(date, country, category)
        |
order_date  country  category   total_revenue  orders  units
2026-01-15  India    Books           1,240.00      18     35
2026-01-15  India    Toys              860.50      11     19

Step 5: Check Quality Before Loading

rows = report.count()
negative = report.filter(F.col("total_revenue") < 0).count()
unknown_share = enriched.filter(F.col("country") == "Unknown").count() / enriched.count()

assert rows > 0, "Report is empty"
assert negative == 0, "Negative revenue found"
assert unknown_share < 0.05, "Too many orders without a matching customer"

A failed check stops the job before bad numbers reach any dashboard.

Step 6: Load the Result

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

(report.write
    .mode("overwrite")
    .partitionBy("order_date")
    .parquet("/data/report/daily_revenue"))

Dynamic overwrite replaces only the dates present in the new data. A rerun of the same day gives identical output.

Project Layout

daily_revenue/
 |-- job.py            main script
 |-- schemas.py        schema definitions
 |-- transforms.py     clean, join, aggregate functions
 |-- tests/            unit tests with small sample data
 |-- job.conf          Spark settings

Running the Job

spark-submit --master yarn --deploy-mode cluster --properties-file job.conf job.py

Ideas to Extend the Project

• Read the input from cloud storage and write the report to a Delta table.
• Add a window function to rank top products per country.
• Stream new orders from Kafka and update the report every few minutes.
• Schedule the job each morning and send an alert when a quality check fails.

Handling Late and Repeated Files

Source systems resend files and deliver some late. Two rules keep the pipeline steady. Track the processed file names or dates in a small control table, and make every run overwrite its own date partition. A resent file then replaces its earlier result and never doubles the totals.

Day 1 file arrives  --> report for day 1 written
Day 1 file resent   --> report for day 1 overwritten with the same values