Broken job: driver OOM from .collect()

PySpark data engineering interview problem. Difficulty: intermediate. Pattern: Driver Memory. About 20 minutes. Part of the Pro drill bank.

The silver events job worked fine on a test sample of 5,000 rows. On Friday it was run on a replay of 40 million rows, and the Spark driver (the one machine that coordinates the job) ran out of memory and was killed. The cause is one line: .collect() pulls every row from the whole cluster into the memory of that single machine.

The shipped script reads ecommerce_events, calls .collect() and builds a new DataFrame from the list. That does no cleanup at all, and on large data it crashes the driver. Rewrite the script so it never collects. Read the table with spark.table("ecommerce_events") and keep the work inside Spark. The same event can be written more than once, so keep 1 row for each event_id, the one with the latest ingest_time. Return event_id, user_id, session_id, event_type, product_id, device, country and event_time, ordered by event_id. Store the DataFrame in result. Do not call .collect() or .toPandas().

Requirements

• Do not use .collect() or .toPandas() anywhere.
• Keep exactly 1 row for each event_id, the row with the latest ingest_time.
• Return the 8 columns listed, without ingest_time.
• Order by event_id and store the DataFrame in result.

Constraints

• The input is the table ecommerce_events, read with spark.table.
• The result stays a Spark DataFrame. It is never turned into a Python list.
• Use a window with row_number() partitioned by event_id and ordered by ingest_time descending.

Examples

Input: event_id | user_id | ingest_time E00013 | U053 | 2026-08-07 09:10:08 E00013 | U053 | 2026-08-07 09:29:08 Output: -- rows.collect() + spark.createDataFrame(rows) is a pure pass-through: it does -- no dedup at all, so both replay rows for E00013 would still be there. -- In production this line never finishes: collecting 40M rows OOMs the -- driver (container killed, code 137) before any output is produced. The bug is not just the OOM: even ignoring memory, .collect() followed by createDataFrame() performs no deduplication, so it can't satisfy the one-row-per-event_id contract either.

Topics: pyspark, OOM, collect.

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