Incremental upsert

PySpark data engineering interview problem. Difficulty: advanced. Pattern: Medallion. About 22 minutes. Part of the Pro drill bank.

A daily job receives a batch of changed customer rows and must apply them to the main customer table. This is called an upsert (update or insert). In Delta Lake you would use MERGE. Here you do the same thing with normal DataFrame steps.

target holds the current customers. batch holds incoming rows. The same customer_id can appear more than once in the batch with different updated_at values, and only the latest version of each customer counts. Return the target after the batch is applied. A customer in the batch takes the latest batch row (the whole row). A customer only in the batch is added. A customer only in the target stays as it is. Return customer_id, name, plan, updated_at, sorted by customer_id. The editor creates both tables and ends with result = target. Change the last line to build your answer. There is no MERGE here, so build the result with a window, an anti join and a union.

Requirements

• Keep only the latest row of each customer_id in batch (by updated_at).
• Remove from target the customers that appear in the batch.
• Combine the untouched target rows with the latest batch rows.
• Return customer_id, name, plan, updated_at, sorted by customer_id, in result.

Constraints

• Use row_number() over a window by customer_id ordered by updated_at descending.
• Use a left anti join to find the target rows that are not in the batch.
• Both parts of the union have the same columns, so unionByName is safe.

Examples

Input: target customer_id | name | plan | updated_at 1 | Ann | basic | 2024-01-01 2 | Bo | pro | 2024-01-01 3 | Cy | basic | 2024-01-01 batch customer_id | name | plan | updated_at 2 | Bo | enterprise | 2024-02-01 4 | Di | basic | 2024-02-01 4 | Di | pro | 2024-02-05 3 | Cy | basic | 2024-02-01 Output: customer_id | name | plan | updated_at 1 | Ann | basic | 2024-01-01 2 | Bo | enterprise | 2024-02-01 3 | Cy | basic | 2024-02-01 4 | Di | pro | 2024-02-05 Customer 1 is untouched, 2 becomes enterprise, 3 is replaced by its batch row (new updated_at), and 4 is inserted with its latest version (pro).

Topics: lakebench, pyspark, upsert, left_anti, row_number, incremental.

More PySpark interview questions · All interview problems · Learn data engineering