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

To keep the newest version of each row in PySpark, number each key's rows from newest to oldest with row_number() over a window partitioned by the key and ordered by the update time, descending, then keep the rows numbered 1. Order by the column that changes when the row changes, such as updated_at, so that 2 versions of a key never tie.

- Page: https://tecyfy.com/data-engineering/course/pyspark-dataframes/keep-the-newest-row
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 2: PySpark DataFrames, Keep the newest row
- 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:

- Keep 1 current row per key from a file of events, with a window and row_number()
- Say why row_number() and not rank(), and why the order column must change with the row
- Remove exact repeats with dropDuplicates() and say when it is needed

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

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

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

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

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/pyspark-dataframes/keep-the-newest-row)

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 dimension](https://tecyfy.com/concepts/slowly-changing-dimensions) 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](/data-engineering/course/pyspark-dataframes/join-aggregate-and-write) joins the current orders to the customers and adds them up.

## Check yourself

1. You keep the newest row of each order with row_number() over a window ordered by ordered_at, descending. When an order is shipped, the shop sends a new row with the same ordered_at and a later updated_at. What does the step keep for that order?

   A. Both versions, because row_number() gives tied rows the same number
   B. Nothing, because Spark raises an error when 2 rows tie in a window
   C. 1 of the 2 versions, and which one is not fixed, so the same code can keep placed on one run and shipped on the next
   D. The version that comes first in the file, because Spark keeps the file's order inside a window

### Answers

1. C. row_number() always gives each row its own number, so ties do not keep both rows; rank() does that. When 2 rows tie on the order column, which one row_number() numbers 1 depends on the order Spark meets them in, which can change with the files and the partitions. Order by updated_at, which changes when the row changes, and the tie is gone.

## Questions

### How do I keep the latest record for each key in PySpark?

Make a window with Window.partitionBy("key").orderBy(F.col("updated_at").desc()), add a column with F.row_number().over(window), keep the rows where it is 1 and drop the column. Each key keeps exactly 1 row, its newest. Order by a column that changes whenever the row changes, so 2 versions of a key never tie.

### What is the difference between row_number and rank in PySpark?

row_number() numbers the rows of a window 1, 2, 3 with no repeats, even when rows tie on the order column. rank() gives tied rows the same number and leaves a gap after them, so 2 rows can both be number 1. To keep exactly 1 row per key, use row_number().

### What is the difference between dropDuplicates and distinct in PySpark?

With no columns, dropDuplicates() and distinct() both remove rows that are the same in every column. dropDuplicates also takes a list of columns, such as dropDuplicates(["order_id"]), and keeps 1 row for each value of them, but it does not let you say which row, so it is no way to keep the newest version. Use a window for that.

## Sources

- [Window functions](https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-window.html)
- [pyspark.sql.functions.row_number](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.row_number.html)
- [pyspark.sql.functions.rank](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.rank.html)
- [DataFrame.dropDuplicates](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.dropDuplicates.html)