This is a common PySpark (Window Functions) interview and practice question. Below is the full answer with a clear, step-by-step explanation.

Question

Problem You are given a PySpark DataFrame containing employee salary history. Each employee can have multiple salary records over time. Sample Data +-----------+--------+-------+----------+ |employee_id|name |salary |updated_at| +-----------+--------+-------+----------+ |101 |Alice |60000 |2024-01-10| |101 |Alice |65000 |2024-06-15| |102 |Bob |55000 |2024-03-01| |102 |Bob |58000 |2024-08-20| |103 |Charlie |70000 |2024-02-11| +-----------+--------+-------+----------+ Task Return only the latest record for every employee based on updated_at . Expected Output +-----------+--------+-------+----------+ |employee_id|name |salary |updated_at| +-----------+--------+-------+----------+ |101 |Alice |65000 |2024-06-15| |102 |Bob |58000 |2024-08-20| |103 |Charlie |70000 |2024-02-11| +-----------+--------+-------+----------+

Answer & Explanation

**What the problem is asking**

You have a table that stores every time an employee’s salary changes.
For each employee you only want the most recent row – the one with the highest `updated_at` value.
The result should contain exactly one record per `employee_id` (the latest salary, name and date).

**Why a window function is the cleanest solution**

A window lets you look at all rows that belong to the same employee, sort those rows by the date, and then assign a sequential number (1, 2, 3 …) inside each employee’s “window”.
If you order the window by `updated_at` descending, the row that gets number 1 is the newest record for that employee.
All you then have to do is keep the rows where the window‑row number equals 1.

This approach has a few advantages:

* It works even when an employee has many salary changes.
* It guarantees you get exactly one row per employee, even if two rows share the same date (you can break ties with a secondary column if needed).
* It avoids a two‑step join that you would need if you first computed the max date per employee and then re‑joined to the original table.

**Step‑by‑step method**

1. **Create a window specification**
```python
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number

w = Window.partitionBy("employee_id") # separate groups per employee
.orderBy(col("updated_at").desc()) # newest date first
```

2. **Add a row number inside each window**
```python
df_with_rn = df.withColumn("rn", row_number().over(w))
```
After this step the data looks like:

```
+-----------+-------+------+----------+---+
|employee_id|name |salary|updated_at|rn |
+-----------+-------+------+----------+---+
|101 |Alice |60000 |2024-01-10| 2 |
|101 |Alice |65000 |2024-06-15| 1 |
|102 |Bob |55000 |2024-03-01| 2 |
|102 |Bob |58000 |2024-08-20| 1 |
|103 |Charlie|70000 |2024-02-11| 1 |
+-----------+-------+------+----------+---+
```

Notice that the newest row for each employee has `rn = 1`.

3. **Filter to keep only the newest rows**
```python
latest_df = df_with_rn.filter(col("rn") == 1).drop("rn")
```

4. **Result**
```python
latest_df.show()
```
```
+-----------+-------+------+----------+
|employee_id|name |salary|updated_at|
+-----------+-------+------+----------+
|101 |Alice |65000 |2024-06-15|
|102 |Bob |58000 |2024-08-20|
|103 |Charlie|70000 |2024-02-11|
+-----------+-------+------+----------+
```

**What to watch out for**

* **Date type** – `updated_at` should be a proper date/timestamp column. If it is a string, cast it first (`col("updated_at").cast("date")`) so the ordering works correctly.
* **Ties** – If two rows for the same employee have the exact same `updated_at`, both will get different row numbers based on the internal order Spark chooses. If you need deterministic tie‑breaking, add a secondary sort column (e.g., `salary` descending) in the `orderBy`.
* **Performance** – Window functions are executed in a single shuffle, which is usually faster than a separate `groupBy` + `join` approach for this pattern.

**Alternative (but less concise) ways**

* **Group‑by + join**: Compute `max(updated_at)` per employee, then join back to the original DataFrame on `employee_id` and `updated_at`. This works but requires two passes over the data.
* **`dropDuplicates` after sorting**:
```python
latest_df = (df.orderBy(col("updated_at").desc())
.dropDuplicates(["employee_id"]))
```
This trick relies on Spark keeping the first row it sees for each key after sorting, but it is less explicit than the window‑row‑number method and can be confusing when reading the code.

**Bottom line**

The most readable, reliable, and idiomatic PySpark solution is:

```python
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number

spark = SparkSession.builder.getOrCreate()

# assume df is already created
w = Window.partitionBy("employee_id").orderBy(col("updated_at").desc())
latest_df = (df
.withColumn("rn", row_number().over(w))
.filter(col("rn") == 1)
.drop("rn"))

latest_df.show()
```

This gives you exactly one row per employee – the most recent salary record – matching the expected output. Mastering this pattern will let you solve many “latest‑per‑group” problems in Spark.