Scenario-based PySpark interview questions, with runnable answers

Scenario-based PySpark questions do not ask what a function does. They describe something that went wrong in a pipeline and ask what you would check and change: the job that got slower after a join, the table that has twice the rows it should, the overwrite that deleted last week's data. Interviewers use them because the answer shows whether you have thought about how Spark moves data, not whether you remember an API.

This guide works through eight of those scenarios. Each one has a small DataFrame you can run, the real output, the mistake that causes the problem, and the fix. Everything below ran on PySpark 3.5 in local mode with 4 cores. To keep results small and repeatable the session sets spark.sql.shuffle.partitions to 4 and turns Adaptive Query Execution off; on a real cluster with AQE on, some of the partition numbers will differ, and the guide says where that matters.

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

spark = (SparkSession.builder.master("local[4]")
         .config("spark.sql.shuffle.partitions", "4")
         .config("spark.sql.adaptive.enabled", "false")
         .getOrCreate())

If DataFrames, joins and groupBy are still new, start with the PySpark track. The scenarios assume you can read DataFrame code.

1"Keep the latest record per customer." Why not dropDuplicates?

A CDC feed sends a new row every time a customer's profile changes. The downstream table should hold 1 row per customer: the newest.

updates = spark.createDataFrame([
    ("C1", "Asha", "Pune", "2026-03-01 09:00"),
    ("C1", "Asha", "Mumbai", "2026-03-04 18:30"),
    ("C2", "Ravi", "Pune", "2026-03-02 10:00"),
    ("C3", "Meena", "Coimbatore", "2026-03-03 12:00"),
    ("C3", "Meena", "Madurai", "2026-03-03 12:00"),
], ["customer_id", "name", "city", "updated_at"]).withColumn("updated_at", F.to_timestamp("updated_at"))

updates.dropDuplicates(["customer_id"]).orderBy("customer_id").show(truncate=False)

+-----------+-----+----------+-------------------+
|customer_id|name |city      |updated_at         |
+-----------+-----+----------+-------------------+
|C1         |Asha |Pune      |2026-03-01 09:00:00|
|C2         |Ravi |Pune      |2026-03-02 10:00:00|
|C3         |Meena|Coimbatore|2026-03-03 12:00:00|
+-----------+-----+----------+-------------------+

dropDuplicates returned 1 row per customer, and for Asha it kept the old Pune address. That is not a bug in Spark: dropDuplicates promises 1 row per key, not which one. In this run it kept the first row it met; on a cluster, with the data split across partitions, it can keep any of them.

The answer interviewers expect is a window ordered by the timestamp, with row_number:

w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc(), F.col("city"))
latest = updates.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")
latest.orderBy("customer_id").show(truncate=False)

+-----------+-----+----------+-------------------+
|customer_id|name |city      |updated_at         |
+-----------+-----+----------+-------------------+
|C1         |Asha |Mumbai    |2026-03-04 18:30:00|
|C2         |Ravi |Pune      |2026-03-02 10:00:00|
|C3         |Meena|Coimbatore|2026-03-03 12:00:00|
+-----------+-----+----------+-------------------+

Asha is now in Mumbai. Meena still needs a decision: her two updates share a timestamp, so the window needs a second sort column or the result is arbitrary again. Here city makes the choice repeatable, but it is not a business rule. In a real feed you would break ties with something that reflects order, like a source log sequence number. Say that in the interview. Practise it: Keep the latest row per key (free), Row Number for Deduplication (free) and Latest status when timestamps tie.

2"The revenue report doubled after we added a join"

An orders table has 3 orders. Someone joins it to the raw updates table from scenario 1 to bring in the customer's city.

orders = spark.createDataFrame([(1, "C1", 1200), (2, "C2", 800), (3, "C3", 2500)],
                               ["order_id", "customer_id", "amount"])
joined = orders.join(updates, "customer_id")
print("orders:", orders.count())
print("after join with raw updates:", joined.count())
print("sum(amount) after join:", joined.agg(F.sum("amount")).first()[0],
      "vs real", orders.agg(F.sum("amount")).first()[0])
print("after join with latest:", orders.join(latest, "customer_id").count())

orders: 3
after join with raw updates: 5
sum(amount) after join: 8200 vs real 4500
after join with latest: 3

C1 and C3 each have 2 rows in updates, so each of their orders is repeated twice. Nothing fails and no error appears; the sum is just wrong. This is called join fan-out, and it is a grain problem: the right side of the join was not 1 row per customer_id.

What to say: check the grain of both sides before the join (groupBy(key).count().filter("count > 1")), deduplicate the dimension to 1 row per key first, and compare row counts before and after the join as a pipeline check. Practise it: Merge DataFrames with Different Join Types (free).

3"A join with a small lookup table is slow"

Joining a large fact table to a small lookup table with a regular join makes Spark shuffle both sides so matching keys meet in the same partition. If one side is small enough to copy to every executor, Spark can skip shuffling the big side entirely. That is a broadcast join.

cities = spark.createDataFrame([("Pune", "West"), ("Mumbai", "West"),
                                ("Coimbatore", "South"), ("Madurai", "South")], ["city", "region"])
big = spark.range(0, 200000).withColumn(
    "city", F.element_at(F.array(*[F.lit(c) for c in ["Pune", "Mumbai", "Coimbatore", "Madurai"]]),
                         (F.col("id") % 4 + 1).cast("int")))

print("autoBroadcastJoinThreshold =", spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
plan_off = big.join(cities, "city")._jdf.queryExecution().executedPlan().toString()
print("auto broadcast off ->", "SortMergeJoin" if "SortMergeJoin" in plan_off else plan_off[:200])
plan_hint = big.join(F.broadcast(cities), "city")._jdf.queryExecution().executedPlan().toString()
print("with F.broadcast hint ->", "BroadcastHashJoin" if "BroadcastHashJoin" in plan_hint else plan_hint[:200])
spark.conf.unset("spark.sql.autoBroadcastJoinThreshold")

autoBroadcastJoinThreshold = 10485760b
auto broadcast off -> SortMergeJoin
with F.broadcast hint -> BroadcastHashJoin

The default threshold is 10 MB (10485760 bytes): Spark broadcasts a side automatically when its estimated size is below that. The example turns automatic broadcasting off to show the two plans side by side; with it on, Spark would already broadcast a 4-row table. The reading habit is the useful part. Call explain() on your join and look for BroadcastHashJoin versus SortMergeJoin.

When does automatic broadcasting fail you? When Spark's size estimate is wrong, typically after filters or on sources without statistics, or when the "small" table is the result of a heavy transformation. Then a F.broadcast() hint fixes it. The opposite mistake is broadcasting something large: every executor must hold the full copy in memory, and the driver must collect it first, so a broadcast of a multi-gigabyte table can fail with out-of-memory errors. Practise it: Broadcast Join (free), and see how Spark chooses shuffle partitions.

4"One task runs for an hour while the rest finish in a minute"

That symptom, one straggler task in a stage, usually means data skew: a few key values hold most of the rows, and every row with the same key goes to the same partition. Here 1,000 events come from India and 10 from each of four other countries.

rows = [(i, "IN") for i in range(1000)] + [(1000 + i, c) for i, c in enumerate(["US", "UK", "SG", "AE"] * 10)]
events = spark.createDataFrame(rows, ["event_id", "country"])
events.groupBy("country").count().orderBy(F.desc("count"), "country").show()

+-------+-----+
|country|count|
+-------+-----+
|     IN| 1000|
|     AE|   10|
|     SG|   10|
|     UK|   10|
|     US|   10|
+-------+-----+

Count rows by key first. It is the cheapest skew check there is. Now see where those rows land when Spark partitions by country, which is what a join or aggregation on country does:

(events.repartition(4, "country")
   .groupBy(F.spark_partition_id().alias("partition")).count().orderBy("partition").show())

+---------+-----+
|partition|count|
+---------+-----+
|        0|   20|
|        1| 1010|
|        2|   10|
+---------+-----+

Partition 1 holds 1,010 of 1,040 rows and partition 3 received nothing, so it does not appear. 3 cores would sit idle while one works. The classic fix is salting: add a small random-looking number to the hot key so its rows spread over several partitions, aggregate on the salted key, then aggregate again without the salt.

SALT = 4
salted = events.withColumn("salt", F.abs(F.hash("event_id")) % SALT)
(salted.repartition(4, "country", "salt")
   .groupBy(F.spark_partition_id().alias("partition")).count().orderBy("partition").show())

+---------+-----+
|partition|count|
+---------+-----+
|        0|  477|
|        1|  275|
|        2|    3|
|        3|  285|
+---------+-----+

The largest partition went from 1,010 rows to 477. The spread is not perfectly even because several (country, salt) pairs can hash to the same partition; with more salt values and more partitions it evens out further. The second aggregation removes the salt and gives the correct totals:

partial = salted.groupBy("country", "salt").count()
final = partial.groupBy("country").agg(F.sum("count").alias("events"))
final.orderBy(F.desc("events"), "country").show()

+-------+------+
|country|events|
+-------+------+
|     IN|  1000|
|     AE|    10|
|     SG|    10|
|     UK|    10|
|     US|    10|
+-------+------+

The salt here comes from hash(event_id) rather than rand() so the result is repeatable when a task is retried. For a skewed join, salting is more work: the large side gets a salt column, and the small side is copied once per salt value so every salted key still finds its match. Also mention that Spark 3's AQE can split skewed join partitions on its own when spark.sql.adaptive.skewJoin.enabled is on, so on a recent cluster check whether that already happens before hand-writing salt. Practise it: Identify Skew and Fix Skew with Salting, and read the step-by-step salting answer.

5"Orders with no promo code vanished from the report"

A left join keeps every order, but the orders without a promo code got no discount, even though the promo table has a row for "no code" with discount 0.

left = spark.createDataFrame([(1, "P1"), (2, None), (3, "P3")], "order_id INT, promo_code STRING")
right = spark.createDataFrame([("P1", 100), (None, 0)], "promo_code STRING, discount INT")
left.join(right, left.promo_code == right.promo_code, "left") \
    .select("order_id", left.promo_code, "discount").orderBy("order_id").show()

+--------+----------+--------+
|order_id|promo_code|discount|
+--------+----------+--------+
|       1|        P1|     100|
|       2|      NULL|    NULL|
|       3|        P3|    NULL|
+--------+----------+--------+

In SQL, and in Spark, NULL = NULL is not true; it is unknown, so the join condition never matches a NULL key. Order 2 survives the left join but gets no discount. If NULL is meant to match NULL, use the null-safe comparison:

left.join(right, left.promo_code.eqNullSafe(right.promo_code), "left") \
    .select("order_id", left.promo_code, "discount").orderBy("order_id").show()

+--------+----------+--------+
|order_id|promo_code|discount|
+--------+----------+--------+
|       1|        P1|     100|
|       2|      NULL|       0|
|       3|        P3|    NULL|
+--------+----------+--------+

Often the better fix is upstream: replace NULL with an explicit value like 'NONE' before the join, so the meaning is in the data rather than in the join condition. There is a performance side too. Many NULL join keys all hash to the same partition, so NULL keys are a common hidden cause of the skew in scenario 4. Practise it: Handle NULL Values (free) and Null count for every column (free).

6"Our daily job writes thousands of tiny files"

Writing with partitionBy creates one folder per partition value, and every Spark task writes its own file into every folder it has rows for. With 8 tasks and 5 dates, that can mean 8 × 5 files.

import glob

sales = spark.range(0, 40000).withColumn(
    "order_date", F.element_at(F.array(*[F.lit(f"2026-03-0{d}") for d in range(1, 6)]),
                               (F.col("id") % 5 + 1).cast("int"))).repartition(8)
print("input partitions:", sales.rdd.getNumPartitions())
sales.write.mode("overwrite").partitionBy("order_date").parquet("/tmp/lakebench_guide/a")
print("files written, no repartition:", len(glob.glob("/tmp/lakebench_guide/a/*/*.parquet")))
sales.repartition("order_date").write.mode("overwrite").partitionBy("order_date").parquet("/tmp/lakebench_guide/b")
print("files written, repartition('order_date'):", len(glob.glob("/tmp/lakebench_guide/b/*/*.parquet")))
print("coalesce(2) partitions:", sales.coalesce(2).rdd.getNumPartitions(),
      "| repartition(16) partitions:", sales.repartition(16).rdd.getNumPartitions())

input partitions: 8
files written, no repartition: 40
files written, repartition('order_date'): 5
coalesce(2) partitions: 2 | repartition(16) partitions: 16

40 files for five dates. On a real pipeline that runs every hour with a few hundred tasks, the count grows fast, and every reader then pays to list and open each small file. Repartitioning by the same column you partition the output by sends each date's rows to one task, so each folder gets 1 file.

The follow-up question is usually coalesce versus repartition. coalesce(n) only merges existing partitions, without a full shuffle, so it is cheap but can only reduce the count and can leave partitions uneven. repartition(n) does a full shuffle, can increase or decrease the count, and spreads rows evenly. 1 file per date is right for a day's 40,000 rows; for a day with billions of rows it would be one enormous file and one slow task, so in practice you pick a target file size and repartition to reach it. Practise it: Small Files Problem and Repartition vs Coalesce.

7"We reran yesterday's job and now the numbers are doubled", or gone

A daily job loads one date into a table partitioned by order_date. The table already holds 4 March. The 5 March load runs, fails half-way through the downstream steps, and someone reruns it. With append:

day = spark.createDataFrame([(1, "2026-03-05", 900), (2, "2026-03-05", 450)], ["order_id", "order_date", "amount"])
other = spark.createDataFrame([(3, "2026-03-04", 300)], ["order_id", "order_date", "amount"])
tgt = "/tmp/lakebench_guide/fact"
other.write.mode("overwrite").partitionBy("order_date").parquet(tgt)
day.write.mode("append").partitionBy("order_date").parquet(tgt)
day.write.mode("append").partitionBy("order_date").parquet(tgt)
print("append twice -> rows:", spark.read.parquet(tgt).count())

append twice -> rows: 5

1 row for 4 March plus 5 March loaded twice: 5 rows where there should be three. The job is not idempotent, meaning running it twice does not give the same result as running it once. The usual fix is to overwrite only the partition you are loading. Spark calls that dynamic partition overwrite:

other.write.mode("overwrite").partitionBy("order_date").parquet(tgt)
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
day.write.mode("overwrite").partitionBy("order_date").parquet(tgt)
day.write.mode("overwrite").partitionBy("order_date").parquet(tgt)
spark.read.parquet(tgt).groupBy("order_date").count().orderBy("order_date").show()

+----------+-----+
|order_date|count|
+----------+-----+
|2026-03-04|    1|
|2026-03-05|    2|
+----------+-----+

Run it twice and the table is the same. Now the trap. The default mode is static, and in static mode an overwrite replaces the whole table, not just the partitions in the DataFrame:

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "static")
day.write.mode("overwrite").partitionBy("order_date").parquet(tgt)
print("static overwrite -> partitions left:",
      [r[0] for r in spark.read.parquet(tgt).select("order_date").distinct().orderBy("order_date").collect()])

static overwrite -> partitions left: [datetime.date(2026, 3, 5)]

4 March is gone. This is the answer interviewers hope to hear: say which overwrite mode you rely on, and set it explicitly in the job rather than trusting the cluster default. On Delta Lake or Iceberg tables you would more often use replaceWhere or a MERGE, which also make the rerun safe. Practise it: Incremental upsert and Upsert that is safe to run twice, and read what happens to other partitions when you overwrite.

8"Orders with an empty cart disappeared after explode"

Each order carries an array of items, and the analytics table wants 1 row per item.

carts = spark.createDataFrame([("O1", ["pen", "book"]), ("O2", []), ("O3", None)],
                              "order_id STRING, items ARRAY<STRING>")
print("explode:"); carts.select("order_id", F.explode("items").alias("item")).show()
print("explode_outer:"); carts.select("order_id", F.explode_outer("items").alias("item")).show()

explode:
+--------+----+
|order_id|item|
+--------+----+
|      O1| pen|
|      O1|book|
+--------+----+

explode_outer:
+--------+----+
|order_id|item|
+--------+----+
|      O1| pen|
|      O1|book|
|      O2|NULL|
|      O3|NULL|
+--------+----+

explode produces no rows for an empty or NULL array, so orders O2 and O3 silently drop out. explode_outer keeps them with a NULL item. Which one is right depends on what the table is for: an item-level sales table can drop empty carts, but an order count built on top of it would be short by two. Practise it: Explode Array Column (free) and Flatten nested JSON records.

How to answer any scenario question

The eight scenarios share a shape, and the same 5 steps work for ones you have not seen:

1. 1Restate the symptom in data terms: too many rows, too few rows, one slow task, too many files, wrong values after a rerun.
2. 2Name the Spark mechanism behind it: grain and join fan-out, shuffle and partitioning, broadcast, NULL semantics, write modes.
3. 3Say how you would confirm it before changing anything: row counts by key, explain(), the stage's task durations in the Spark UI, file counts per folder.
4. 4Give the fix and its cost: salting adds a second aggregation, broadcasting costs executor memory, repartition is a shuffle.
5. 5Say how you would stop it coming back: a row-count check after joins, an explicit overwrite mode, a tie-breaker column.

The free PySpark problems on LakeBench run these patterns in the browser, and the theory question bank covers the concepts one question at a time.