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

A PySpark job you can trust keeps its logic in functions that take and return DataFrames, keeps every row it rejects where someone can fix it, and checks its own output before anything reads it. clean_orders.py does that for the course's raw orders: 8 rows read, 1 rejected, 4 current orders, 1 order with an unknown customer.

- Page: https://tecyfy.com/data-engineering/course/pyspark-dataframes/build-it
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 2: PySpark DataFrames, Build it: clean the raw orders
- Parts of this module: Read files with a schema (https://tecyfy.com/data-engineering/course/pyspark-dataframes); Keep the newest row (https://tecyfy.com/data-engineering/course/pyspark-dataframes/keep-the-newest-row); Join, aggregate and write (https://tecyfy.com/data-engineering/course/pyspark-dataframes/join-aggregate-and-write); Serverless, DuckDB and Polars (https://tecyfy.com/data-engineering/course/pyspark-dataframes/serverless-duckdb-polars); Build it: clean the raw orders (https://tecyfy.com/data-engineering/course/pyspark-dataframes/build-it)

At the end of this part you can:

- Write a PySpark job as small functions that take and return DataFrames
- Keep the rejected rows and the unmatched rows instead of losing them
- Stop a job with checks on its output before a wrong table reaches anyone

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](/data-engineering/course/pyspark-dataframes) made:

```python
"""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`:

```text
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](/data-engineering/course/pyspark-dataframes/join-aggregate-and-write) so it stops.

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

- Quiz: [PySpark Basics](https://tecyfy.com/quiz/pyspark/pyspark-basics). Evaluate your foundational understanding of PySpark with essential concepts and beginner-level tasks.

## Check yourself

1. clean_orders.py checks that its orders have as many rows as distinct order_id values, and stops if not. Which mistake is this check there to catch?

   A. A file with a negative amount in it
   B. A window or a join that copied an order, so every total after it would count the order twice
   C. An order_id that Parquet cannot store
   D. A customer in the customers file with no orders

### Answers

1. B. The step promises 1 row per order. A window ordered by a column that ties, or a join to a key that is not unique, breaks that promise without an error, and every sum after it is too high. Counting rows against distinct keys finds it at once. The negative amount has its own check, and a customer with no orders is not a problem for this step.

## Questions

### How do I check data quality in a PySpark job?

Compute each rule as a small query and stop the job when one fails, before the output is used. Compare count() with a distinct count of the key for duplicates, and use isEmpty() on a filter for rows that must not exist, such as negative amounts. In production, raise an exception rather than use assert, which Python skips when it runs with -O.

### How do I run a PySpark script outside a notebook?

With PySpark installed by pip, run it like any Python program, python clean_orders.py, and the script starts Spark itself with SparkSession.builder.getOrCreate(). On a cluster you hand the same file to spark-submit, or on Databricks to a job, which module 10 covers.

## Sources

- [DataFrame.isEmpty](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.isEmpty.html)
- [The assert statement](https://docs.python.org/3/reference/simple_stmts.html#the-assert-statement)
- [Submitting applications](https://spark.apache.org/docs/latest/submitting-applications.html)