PySpark joins, groupBy and writing Parquet without losing or copying rows

A join is where a pipeline most often goes wrong without telling anyone. It can drop rows, so a total comes out short, or copy them, so a total comes out high, and both look like a normal table. This part joins the current orders from part 2 to the customers, adds them up and writes the result.

Start from the current orders

If you are starting here, this block rebuilds what parts 1 and 2 made, from the bronze copy part 1 saved:

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("orders").getOrCreate()
RAW, LAKE = "raw", "lake"
good = spark.read.parquet(f"{LAKE}/bronze_orders").filter(F.col("_bad").isNull()).drop("_bad")
newest_first = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
current = (
    good.dropDuplicates()
    .withColumn("version", F.row_number().over(newest_first))
    .filter(F.col("version") == 1)
    .drop("version")
)

Join, and count what comes out

Add each customer's name and country. A left join keeps every order, with nulls where no customer matches:

customers = spark.read.csv(
    f"{RAW}/customers.csv", header=True, schema="customer_id STRING, name STRING, country STRING"
)
with_customer = current.join(customers, on="customer_id", how="left")
with_customer.select("order_id", "status", "amount", "name", "country").orderBy("order_id").show()

+--------+---------+------+----+-------+
|order_id|   status|amount|name|country|
+--------+---------+------+----+-------+
|    1001|  shipped| 25.50| Ana|     PT|
|    1002|cancelled| 12.00| Ben|     IE|
|    1004|   placed| 40.00| Ana|     PT|
|    1005|   placed|  8.75|NULL|   NULL|
+--------+---------+------+----+-------+

Order 1005 belongs to customer C09, who is not in the file. An inner join, which is what you get when you leave out how, would have dropped it, and the revenue would be short by 8.75 with nothing to show why. To find such rows on purpose, use a left_anti join. It returns the left rows that have no match, and nothing else:

current.join(customers, on="customer_id", how="left_anti").select("order_id", "customer_id").show()

+--------+-----------+
|order_id|customer_id|
+--------+-----------+
|    1005|        C09|
+--------+-----------+

Joins can also add rows, because a join returns 1 row for every pair of rows whose keys match. If a customer appears twice in the customers file, each of their orders comes out twice. Switch the demo's file to C01 twice and watch Ana's orders double.

current.join(customers, on="customer_id", how="left")

So before you join, check the side that should have 1 row per key: customers.count() == customers.select("customer_id").distinct().count(). Passing on="customer_id" as a name, rather than a condition such as current.customer_id == customers.customer_id, also keeps a single customer_id column in the result instead of 2 with the same name.

Add it up

Revenue per day and country, leaving out cancelled orders:

daily = (
    with_customer.filter(F.col("status") != "cancelled")
    .groupBy(F.to_date("ordered_at").alias("day"), "country")
    .agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue"))
    .orderBy("day", "country")
)
daily.show()

+----------+-------+------+-------+
|       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|
+----------+-------+------+-------+

Each order counts once, on the day it was made, in its current state. That is the answer the raw file could not give.

If you call daily.explain(), 2 words in the plan are worth knowing now. Exchange is a shuffle: Spark moving rows between machines so that all the rows of 1 order, or 1 day, end up together, which is the costly part of most jobs. BroadcastHashJoin means Spark sent the small customers table to every worker instead of shuffling the orders, which it does by itself for tables under 10 MB. Module 3 reads these plans in the Spark UI and sizes the shuffles.

The same query can be SQL. Name a DataFrame as a temporary view with with_customer.createOrReplaceTempView("orders_now"), and spark.sql("SELECT ... FROM orders_now GROUP BY ...") returns a DataFrame like any other. Both become the same plan on the same engine, so choose whichever reads better for the step.

Write it so it can run again

Writing is an action, so this is where the whole plan runs:

(
    current.withColumn("order_date", F.to_date("ordered_at"))
    .write.mode("overwrite")
    .partitionBy("order_date")
    .parquet(f"{LAKE}/orders")
)

mode("overwrite") replaces what was there, which makes the step safe to run again: run it twice and you get the same result, not the orders twice. That property, called an idempotent pipelineIdempotent Pipelines and Exactly-OnceA pipeline step is idempotent when running it twice gives the same result as running it once. That is what makes retries safe, and it is how systems that deliver a message more than once still count it exactly once., is what lets a scheduler retry a failed run without anyone checking it by hand. The default mode, errorifexists, fails if the folder exists, and append adds to it, which duplicates every row on a rerun.

partitionBy("order_date") writes 1 folder per day, order_date=2026-10-01 and so on, so a query for 1 day reads 1 folder. It helps only when each partition is large. Databricks' own advice is not to partition a table under 1 TB and to give each partition at least 1 GB; with less, thousands of tiny files make every query slower. On Databricks you will use liquid clustering instead, in module 4.