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

A PySpark join returns 1 row for every pair of rows whose keys match. An inner join drops the rows with no match, a left join keeps them with nulls, and a left_anti join returns only them. A key that appears twice on the other side copies the row, so check that side's key is unique before you join, and count the rows after.

- Page: https://tecyfy.com/data-engineering/course/pyspark-dataframes/join-aggregate-and-write
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 2: PySpark DataFrames, Join, aggregate and write
- 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:

- Join DataFrames with inner, left and left_anti, and count the rows each returns
- Find the rows a join did not match, and the keys that copy rows
- Add up a DataFrame with groupBy and agg, in Python or in SQL
- Write the result with a save mode that is safe to run again

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](/data-engineering/course/pyspark-dataframes/keep-the-newest-row) 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:

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

```python
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()
```

```text
+--------+---------+------+----+-------+
|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:

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

```text
+--------+-----------+
|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.

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/pyspark-dataframes/join-aggregate-and-write)

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:

```python
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()
```

```text
+----------+-------+------+-------+
|       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.

- Tool: [Spark shuffle partition calculator](https://tecyfy.com/data-engineering/spark-shuffle-partitions). The data a shuffle moves, your cores and one heavy key in; the partitions, the tasks adaptive query execution makes and the settings out.

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:

```python
(
    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 pipeline](https://tecyfy.com/concepts/idempotent-pipeline), 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.

## Check yourself

1. You join 1,000,000 orders to a customers table on customer_id with a left join, and the result has 1,000,340 rows. Every order has exactly 1 customer_id. What happened?

   A. Some customer_id values appear more than once in the customers table, so those orders were copied once for each match
   B. A left join adds a row for each customer with no orders
   C. Spark split some orders across partitions during the shuffle and counted them twice
   D. The orders with no matching customer were added again at the end with nulls

### Answers

1. A. A join returns 1 row for every pair of rows whose keys match, so a key that appears twice on the right copies the order. A left join keeps every row of the left side once when there is no match, with nulls, and never adds rows from the right side alone. Shuffles move rows but never duplicate them. Check the right side's key is unique before the join, with a count of rows against a count of distinct keys.

## Questions

### What is the difference between an inner join and a left join in PySpark?

An inner join keeps only the rows that found a match on both sides. A left join keeps every row of the left DataFrame, with nulls in the right side's columns where there was no match. inner is the default when you do not pass how, so a row with no match disappears without an error.

### How do I find the rows that did not match in a PySpark join?

Use a left_anti join, left.join(right, on="key", how="left_anti"). It returns the rows of the left DataFrame that have no match on the right, with only the left side's columns. Count them in every pipeline run, so a row that stops matching is seen rather than lost.

### What is the difference between overwrite and append in PySpark write?

mode("overwrite") replaces what is at the path or table, so running the step twice gives the same result. mode("append") adds the new rows to what is there, so a rerun adds every row again. The default, errorifexists, fails if anything is already there.

### Is PySpark the same as Spark SQL?

They are 2 ways of writing the same queries. A DataFrame built with PySpark and the same query written in Spark SQL become the same plan and run on the same engine at the same speed, so you can mix them, for example by naming a DataFrame as a view with createOrReplaceTempView and querying it with spark.sql.

## Sources

- [DataFrame.join](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.join.html)
- [Performance tuning](https://spark.apache.org/docs/latest/sql-performance-tuning.html)
- [Generic load and save functions](https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html)
- [DataFrameWriter.mode](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.mode.html)
- [When to partition tables on Databricks](https://docs.databricks.com/aws/en/tables/partitions)