PySpark: keep the newest version of each row with row_number()

Source systems rarely send you a clean table. They send events: an order was placed, then shipped, then perhaps sent again by a retry. Turning those into 1 row per order, the order as it is now, is the most common job in data engineering, and this part does it with a window function.

Part 1 ended with a wrong answer. Asked for the orders that are still placed, Spark returned 1001 and 1002, which have since been shipped and cancelled, and 1004 twice. Nothing was wrong with the filter. The data holds every version of each order, and the filter matched the old ones.

Start from the saved orders

This part reads the bronze copy part 1 saved. If you are starting here, run part 1's setup and its read and save first, then:

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

spark = SparkSession.builder.appName("orders").getOrCreate()
LAKE = "lake"
good = spark.read.parquet(f"{LAKE}/bronze_orders").filter(F.col("_bad").isNull()).drop("_bad")

On Databricks, read the table instead: spark.table(f"{catalog}.{schema}.bronze_orders"), with the same filter.

Number the versions, keep the newest

For each order, number its versions from newest to oldest and keep number 1. dropDuplicates() first removes exact repeats, such as the 2 identical rows of order 1004:

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")
)
current.orderBy("order_id").show()

+--------+-----------+---------+------+--------+-------------------+-------------------+
|order_id|customer_id|   status|amount|currency|         ordered_at|         updated_at|
+--------+-----------+---------+------+--------+-------------------+-------------------+
|    1001|        C01|  shipped| 25.50|     EUR|2026-10-01 09:15:00|2026-10-02 08:00:00|
|    1002|        C02|cancelled| 12.00|     EUR|2026-10-01 10:02:00|2026-10-03 09:00:00|
|    1004|        C01|   placed| 40.00|     EUR|2026-10-02 12:45:00|2026-10-02 12:45:00|
|    1005|        C09|   placed|  8.75|     EUR|2026-10-03 14:10:00|2026-10-03 14:10:00|
+--------+-----------+---------+------+--------+-------------------+-------------------+

4 rows, 1 per order, each in its current state. A window is a group of rows that a function looks across without collapsing them, unlike groupBy, which returns 1 row per group. partitionBy("order_id") makes 1 window per order, orderBy sorts inside it, and row_number() counts 1, 2, 3 down the sorted rows.

2 choices decide the answer

The first is the function. row_number() gives every row its own number, even when 2 rows tie on the order column. rank() gives tied rows the same number, so when 2 versions share a time, both are number 1 and you keep both. To keep exactly 1 row per key, use row_number().

The second is the column you order by: updated_at, when the row changed, not ordered_at, when the order was made. An order keeps its ordered_at when its status changes, so ordered by that column the placed and shipped versions of 1001 tie. row_number() still numbers one of them 1, but which one depends on the order Spark meets the rows in, and that can change with the files and how they are split. The same code can then report a shipped order as placed on one run and get it right on the next, which is the worst kind of wrong: it passes a test once.

Try both in the demo, and the repeats too.

newest_first = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
good.dropDuplicates().withColumn("version", F.row_number().over(newest_first))
    .filter(F.col("version") == 1)

With row_number() the window keeps 1 of the identical 1004 rows even without dropDuplicates(), so here it changes nothing. Keep it anyway: it costs little, and it protects the next person who switches to rank(). Do not reach for dropDuplicates(["order_id"]) to keep 1 row per order, either. It keeps 1 row for each order, but not one you can choose.

This pattern, 1 current row per key that is overwritten as the key changes, is the heart of slowly changing dimensionSlowly Changing DimensionsThe ways a warehouse handles a change to a descriptive record, such as a customer's address: type 1 overwrites it, type 2 adds a new dated version, and type 3 keeps the previous value in an extra column. type 1, which module 1 covers in SQL. Type 2, which keeps every version with the dates it was true, builds on the same window. Part 3 joins the current orders to the customers and adds them up.