Build a PySpark job that cleans raw orders, with data checks

The course's project is one shop's order pipeline, from raw files to a dashboard and an AI answer. This part adds its first step. Everything in it comes from parts 1 to 3, put into 1 script you can run, schedule and test: read the raw orders, keep what does not parse, keep 1 current row per order, and check the result before anyone reads it.

Build it: clean the raw orders

Save this as clean_orders.py, next to the raw folder that part 1 made:

"""Clean the raw orders: the first step of the course's order pipeline."""
from pyspark.sql import DataFrame, SparkSession, Window
from pyspark.sql import functions as F

RAW = "raw"    # where the raw files are
LAKE = "lake"  # where this step writes

ORDERS = """
    order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2),
    currency STRING, ordered_at TIMESTAMP, updated_at TIMESTAMP, _bad STRING
"""
CUSTOMERS = "customer_id STRING, name STRING, country STRING"


def read_orders(spark: SparkSession, path: str) -> DataFrame:
    """Every row of the file, with the rows that do not parse kept in _bad."""
    return (
        spark.read.schema(ORDERS)
        .option("header", True)
        .option("mode", "PERMISSIVE")
        .option("columnNameOfCorruptRecord", "_bad")
        .csv(path)
    )


def current_orders(orders: DataFrame) -> DataFrame:
    """One row per order: its newest version, with exact repeats removed first."""
    newest_first = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
    return (
        orders.dropDuplicates()
        .withColumn("version", F.row_number().over(newest_first))
        .filter(F.col("version") == 1)
        .drop("version")
        .withColumn("order_date", F.to_date("ordered_at"))
    )


def daily_revenue(orders: DataFrame, customers: DataFrame) -> DataFrame:
    """Orders and revenue per day and customer country, cancelled orders left out."""
    return (
        orders.join(customers, on="customer_id", how="left")
        .filter(F.col("status") != "cancelled")
        .groupBy(F.col("order_date").alias("day"), "country")
        .agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue"))
    )


def main(spark: SparkSession) -> None:
    read_orders(spark, f"{RAW}/orders.csv").write.mode("overwrite").parquet(f"{LAKE}/bronze_orders")
    bronze = spark.read.parquet(f"{LAKE}/bronze_orders")

    bad = bronze.filter(F.col("_bad").isNotNull())
    good = bronze.filter(F.col("_bad").isNull()).drop("_bad")
    bad.write.mode("overwrite").parquet(f"{LAKE}/rejected_orders")

    orders = current_orders(good)
    orders.write.mode("overwrite").partitionBy("order_date").parquet(f"{LAKE}/orders")

    customers = spark.read.csv(f"{RAW}/customers.csv", header=True, schema=CUSTOMERS)
    unknown = orders.join(customers, on="customer_id", how="left_anti")

    # Checks: stop the step if the output is wrong, before anyone reads it.
    assert orders.count() == orders.select("order_id").distinct().count(), "an order appears twice"
    assert orders.filter(F.col("amount") < 0).isEmpty(), "a negative amount"
    print(f"rows read: {bronze.count()}, rejected: {bad.count()}, orders: {orders.count()}, unknown customers: {unknown.count()}")

    daily_revenue(orders, customers).orderBy("day", "country").show()


if __name__ == "__main__":
    main(SparkSession.builder.appName("clean-orders").getOrCreate())

Run it with python clean_orders.py:

rows read: 8, rejected: 1, orders: 4, unknown customers: 1
+----------+-------+------+-------+
|       day|country|orders|revenue|
+----------+-------+------+-------+
|2026-10-01|     PT|     1|  25.50|
|2026-10-02|     PT|     1|  40.00|
|2026-10-03|   NULL|     1|   8.75|
+----------+-------+------+-------+

3 choices in it carry through the rest of the course.

The logic is in functions that take and return DataFrames. read_orders, current_orders and daily_revenue read no paths of their own and write nothing, so module 11 can test each one with a few rows made in the test and no files. main is the only part that knows where the data lives.

Nothing is thrown away. The rejected row goes to its own folder with the exact line that failed, and the order with an unknown customer is counted on every run. Both are for someone to fix at the source, and a count that grows from 1 to 400 overnight tells you a source changed before a customer does.

The checks run before the output is used. The first says each order appears once, which catches a window ordered by a column that ties and a join to a key that is not unique. The second uses isEmpty(), which stops at the first matching row instead of counting them all. A broken file stops the step instead of reaching a dashboard. assert keeps the script short; Python skips it when run with -O, so a production job raises an exception instead, and module 7 moves these rules into the pipeline itself.

On Databricks, set RAW to your volume's path and write tables in place of the Parquet folders: .saveAsTable(f"{catalog}.{schema}.bronze_orders") for the first write, spark.table(...) to read it back, and the same for the rejected rows and the orders, without partitionBy.

Then try 2 changes on your own. Add an order with a negative amount to the file and watch the check stop the run. Then add a second row for customer C01 to the customers file and see what the join does to the revenue: the script still runs, because nothing checks the customers' key. Add the unique-key check from part 3 so it stops.

Next, module 3 opens the Spark UI for this script and shows where its time goes.

QuizPySpark BasicsEvaluate your foundational understanding of PySpark with essential concepts and beginner-level tasks.