Write the job. Marked on whether you stayed in the DataFrame API where the optimizer can see you, and on whether the shape of the output is what was asked for.
Top-N per group, deduplication, pivots and the grain of what you emit. Where most PySpark answers are wrong before performance is even discussed.
Staying where the optimizer can see you
5
Native expressions against UDFs, and the higher-order functions that cover far more than people reach for.
Writing joins that behave
5
Column ambiguity, join types, null keys and the fan-out that silently multiplies your fact table.
Reading and writing
5
Schemas, partitioned writes, file counts and the modes that decide whether a rerun is safe.
Evergreen · asked verbatim
2
The flat form, in the words interviewers actually use. Same ground as the questions above, asked as recall rather than as a situation — because the two fail separately, and a candidate who can read a plan can still stall on “define a stage”.
01 / 22
Wide vs narrow transformationsShuffle mechanics
Return the two largest orders per region. Write it, and say what makes the result deterministic when two orders tie on amount.
from pyspark.sql.window import Window
w = Window.partitionBy("region").orderBy(F.col("amount").desc(), F.col("order_id"))
out = (spark.table("orders")
.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") <= 2)
.select("region", "order_id", "amount"))
The code — predict the output before reading on
from pyspark.sql.window import Window
w = Window.partitionBy("region").orderBy(F.col("amount").desc(), F.col("order_id"))
out = (spark.table("orders")
.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") <= 2)
.select("region", "order_id", "amount"))
Why they ask this
Top-N per group is the most-asked PySpark coding question, and the tie-breaking half is what separates a correct answer from one that passes on the sample data.
Say this
row_number over a window partitioned by region and ordered by amount descending, filtered to rn <= 2. Determinism comes from adding a unique tiebreaker to the ORDER BY — without it, ties resolve arbitrarily and the answer changes between runs.
The reasoning
The window form is the standard answer: partition by the group, order by the measure, assign row_number, and filter. The plan shows what it costs — an Exchange to bring each region together and a Sort within the partition — and the filter on rn cannot be pushed below the window, because rn does not exist until the window has run.
The determinism question is the real content. row_number assigns 1, 2, 3 with no ties, so if two orders share the top amount, which one gets rank 1 is decided by whatever order the sort happened to produce. Add a unique column — order_id here — as a secondary sort key and the result is stable across runs, which matters the first time someone diffs two runs of the same job.
Know the three ranking functions and when each is right, because interviewers ask. row_number gives 1,2,3,4 and is what you want for 'exactly two'. rank gives 1,2,2,4 and returns more than N rows when there are ties at the boundary — which is correct when the requirement is 'everyone tied for second'. dense_rank gives 1,2,2,3. Choosing rank when the requirement said exactly two is a common and invisible bug.
What it actually returns 1 shuffle, run on Spark 4.2
Exchange plus Sort. The order_id tiebreaker is what makes it reproducible.
"Use orderBy then limit per group." limit applies to the whole DataFrame, not per group. Getting per-group behaviour without a window means a self-join against a per-group max, which is more code and more shuffles.
They’ll ask next
The requirement changes to 'everyone tied for second place too'. Which function?
from pyspark.sql.window import Window
w = Window.partitionBy("customer_id").orderBy(F.col("order_date").desc(), F.col("order_id").desc())
out = (spark.table("orders")
.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") == 1)
.select("customer_id", "order_id", "order_date", "amount")
.filter(F.col("customer_id") < 4))
The code — predict the output before reading on
from pyspark.sql.window import Window
w = Window.partitionBy("customer_id").orderBy(F.col("order_date").desc(), F.col("order_id").desc())
out = (spark.table("orders")
.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") == 1)
.select("customer_id", "order_id", "order_date", "amount")
.filter(F.col("customer_id") < 4))
Why they ask this
Deduplication appears in nearly every pipeline, and dropDuplicates is the wrong tool that looks like the right one.
Say this
A window ordered by date descending with row_number = 1. dropDuplicates keeps an *arbitrary* row per key, not the latest — it has no ordering, so which row survives is whatever arrived first in an unspecified order.
The reasoning
dropDuplicates(['customer_id']) does keep one row per customer. It has no way to know which one you want, so it keeps whichever it encounters first, and that depends on partitioning and task completion order. It is not deterministic and it is not 'latest' — on a rerun you can get a different row.
The window form states the intent: partition by customer, order by date descending with a unique tiebreaker, keep row_number = 1. It costs a shuffle and a sort, which is the honest price of an ordered choice.
dropDuplicates is right when the duplicates are genuinely identical — the at-least-once redelivery case, where you want any one of N byte-identical copies. Naming that distinction is what an interviewer is listening for: dropDuplicates for *identical* rows, a window for *choosing among differing* rows. Using the first where you meant the second is a silent correctness bug that only shows up when someone compares two runs.
What it actually returns 1 shuffle, run on Spark 4.2
Ordered by date then order_id, so the surviving row is the same every run.
"dropDuplicates keeps the first row, so sort first and it works." A sort before dropDuplicates is not preserved through the shuffle it performs. There is no documented ordering guarantee to rely on, and relying on it is how this bug ships.
They’ll ask next
Ten million customers and a hundred rows each. Is the window still the right answer?
pivot is a genuinely useful operator most people do not know, and its two-pass behaviour without an explicit value list is a real production cost.
Say this
groupBy the row key, pivot on the column, aggregate. Passing the values explicitly matters because without them Spark runs an extra job to discover the distinct values first — and the output schema becomes data-dependent.
The reasoning
The operator is groupBy(row_key).pivot(column, values).agg(...), and the plan shows the aggregation shape you would expect. It is far better than hand-writing a sum(when(region=='EU', amount)) per region, which is what people reach for.
Passing the values explicitly does two things. It avoids an extra job: without the list Spark has to compute the distinct values before it can plan the pivot, which means a separate action on the driver before your query runs. And it fixes the output schema — with inference, a region appearing or disappearing in the data silently changes your columns, which breaks whatever reads the result.
The limit worth naming is cardinality. A pivot produces one column per value, so pivoting on something with thousands of distinct values produces a table with thousands of columns, and Spark will refuse past a configured maximum. If the answer to 'how many values' is 'it varies', a pivot is the wrong shape and the long format with a filter downstream is the right one.
What it actually returns 2 shuffles, run on Spark 4.2
Values passed explicitly, so the schema is fixed and there is no discovery job.
"Write a sum(when(...)) per value — it is the same thing." It is, and it does not scale past a handful of values and has to be edited when one is added. pivot expresses the intent and Spark generates the same aggregation.
They’ll ask next
You pivot on product_id and there are 40,000 of them. What happens?
UDFs vs native vs pandas UDFsPredicate & projection pushdown
Sum only the orders over 300, and count them, in one pass per region. Write it natively, and compare against the UDF version somebody actually shipped.
big = F.udf(lambda a: a if a > 300 else 0, "int")
out = (spark.table("orders")
.withColumn("big", big(F.col("amount")))
.repartition(40, "customer_id")
.groupBy("region")
.agg(F.sum("big").alias("big_total"),
F.count(F.when(F.col("amount") > 300, True)).alias("big_n")))
Conditional aggregation is the single most common place a UDF gets written unnecessarily, and the native form is not obvious to people who have not seen it.
Say this
F.sum(F.when(cond, col).otherwise(0)) and F.count(F.when(cond, True)) — both compile into the aggregate. The UDF version returns the same rows and adds a BatchEvalPython node and a pointless shuffle.
The reasoning
when/otherwise inside an aggregate is the idiom, and it is worth committing to memory because it replaces a large fraction of the UDFs people write. Several conditional measures can go in one agg() call, so the data is scanned and grouped once regardless of how many conditions you need.
count(when(cond, True)) is the second half and the part people miss: count ignores nulls, so when-without-otherwise yields null for non-matching rows and they are simply not counted. That is a neat, idiomatic conditional count with no filter and no second aggregation.
The two versions below return identical rows — the harness asserts it — and differ in the plan. The UDF version adds Python evaluation and, in the form it was actually written, a repartition that buys nothing. This is the shape of most real PySpark review comments: the code works, and it is doing several times the necessary work for a result the DataFrame API expresses directly.
What it actually returns 1 shuffle, run on Spark 4.2
Identical rows either way. One version leaves the JVM and shuffles for nothing.
The physical plan1 shuffle
== Physical Plan ==
*(2) HashAggregate(keys=[region#1], functions=[sum(CASE WHEN (amount#2 > 300) THEN amount#2 ELSE 0 END), count(CASE WHEN (amount#2 > 300) THEN true END)])
+- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
+- *(1) HashAggregate(keys=[region#1], functions=[partial_sum(CASE WHEN (amount#2 > 300) THEN amount#2 ELSE 0 END), partial_count(CASE WHEN (amount#2 > 300) THEN true END)])
+- *(1) ColumnarToRow
+- BatchScan parquet /tables/orders[region#1, amount#2] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string,amount:int> RuntimeFilters: []
It returns
region
big_total
big_n
APAC
178624
365
EU
179217
366
US
179133
366
The answer most people give
"Filter first, then aggregate, then join the two results together." That reads the data twice and adds a join to reassemble what one pass produced. when/otherwise gets both measures from a single scan.
They’ll ask next
You need the same measure over five different thresholds. Does your approach still hold?
Join strategies (broadcast / SMJ / skew join / salting)RDD vs DataFrame vs Dataset
Both tables have customer_id. The join runs and then select('customer_id') raises AnalysisException. Why, and what is the fix that keeps working after the next column is added?
customers(customer_id int, tier string, country string)
400 customers. Small enough to broadcast.
customers
customer_id
tier
country
0
free
EU
1
pro
US
2
free
APAC
3
pro
EU
The code — what does Spark do with it?
o = spark.table("orders").alias("o")
c = spark.table("customers").alias("c")
out = (o.join(c, F.col("o.customer_id") == F.col("c.customer_id"), "left")
.select(F.col("o.order_id"), F.col("o.amount"), F.col("c.tier"))
.filter(F.col("o.order_id") < 5))
The code — predict the output before reading on
o = spark.table("orders").alias("o")
c = spark.table("customers").alias("c")
out = (o.join(c, F.col("o.customer_id") == F.col("c.customer_id"), "left")
.select(F.col("o.order_id"), F.col("o.amount"), F.col("c.tier"))
.filter(F.col("o.order_id") < 5))
Why they ask this
Column ambiguity is the most common PySpark error, and the fix people reach for — dropping one side — breaks quietly when schemas change.
Say this
Both sides contributed a column of that name, so the reference is ambiguous. Alias each side and qualify every column you select — that keeps working when either schema gains a column.
The reasoning
Joining on an equality expression keeps both key columns in the output, so the name resolves to two attributes and analysis fails. Using the string form — join(c, 'customer_id') — coalesces the key into one column, which is why that form is preferred when the names match on both sides and you want the key once.
When you need the expression form, alias each DataFrame and qualify your selects. That is explicit about which side every column comes from, and it survives either table gaining a column with a name the other already has — which is exactly the situation where drop('customer_id') starts removing the wrong one or failing.
The subtle version worth mentioning is that a column can be ambiguous even when it looks unique, because Spark tracks attributes by id rather than by name. Self-joins are the classic case: joining a DataFrame to itself gives two attributes with identical names and identical lineage, and the only reliable fix is aliasing both sides before the join.
What it actually returns 0 shuffles, run on Spark 4.2
Aliased sides, qualified selects. Zero shuffles — customers is broadcast.
"Just drop the duplicate column after the join." It works until a second column collides, and on a self-join it can drop the wrong one — the names are identical and Spark resolves by attribute id, not by which one you meant.
They’ll ask next
You need to self-join orders to itself on a lag. Write the join header.
Wide vs narrow transformationsRDD vs DataFrame vs Dataset
Each order row carries an array of item structs. Produce one row per item with the order's fields alongside, and say what happens to an order whose array is empty.
Why they ask this
Nested data is everywhere in event pipelines, and the empty-array case is the bug that silently drops rows.
Say this
explode the array and select through the struct. An order with an empty or null array disappears entirely — explode_outer is what keeps it, with nulls for the item fields.
The reasoning
explode turns one row with an N-element array into N rows, carrying the other columns along. Struct fields are reached with dot notation after exploding, so item.sku and item.qty become ordinary columns. It is a narrow transformation — no shuffle — even though the row count changes.
The trap is that explode drops rows whose array is empty or null, because zero elements means zero output rows. If the requirement is one row per item, that is correct. If it is 'every order, with its items', it silently loses every order that had none — which is a data loss bug that never raises. explode_outer emits one row with null item fields instead, which is the outer-join equivalent.
Worth knowing beside it: posexplode when you need the array index, and the higher-order functions — transform, filter, aggregate over arrays — which let you operate on the array without exploding at all. Exploding to filter and then re-grouping is a common and expensive pattern that filter(array, lambda) replaces with no shuffle.
Two shuffles to do what F.filter(col('items'), lambda x: ...) does with none.
The answer most people give
"explode is a wide transformation because it changes the row count." Row count and data movement are different things. Each output row comes from exactly one input row, so it is narrow and no shuffle is involved.
They’ll ask next
You only want items over 100 units. Where do you filter, and why does it matter?
You are asked for 'revenue by customer and month'. Before writing any code, what do you need to pin down, and what does getting it wrong look like downstream?
Why they ask this
Most wrong PySpark answers are wrong about grain rather than about Spark. An interviewer wants to hear the clarifying question before the code.
Say this
One row per customer per month — and whether months with no orders should appear, which currency, and whether refunds are negative rows or a separate column. Getting it wrong shows up as a join downstream that fans out or drops rows.
The reasoning
The grain is the sentence that says what one row means. 'One row per customer per calendar month, in reporting currency, including months with zero revenue' is answerable; 'revenue by customer and month' is not, and the ambiguity is where the bugs live.
Missing months are the first question because they change the shape of the code entirely. A groupBy produces only months that had orders; if the consumer expects a dense series — for a chart, or a month-over-month calculation — you have to generate the calendar and left join to it. That is a completely different job from the one a groupBy writes.
Getting the grain wrong is invisible in the output and loud downstream. A result at a finer grain than declared fans out when someone joins it, doubling their numbers. A result at a coarser grain drops detail nobody can recover. Both surface as 'the dashboard is wrong' weeks later, which is why stating the grain out loud before writing is the habit being tested.
The formulations
State the grain, then write to itship
-- one row per (customer_id, month), reporting currency,
-- months with no orders present with 0
groupBy('customer_id','month').agg(...)
The grain is written down where the next reader will find it, and the code is checkable against it.
Sparse by construction. Fine if that is what was wanted, and nobody asked.
The answer most people give
"Just group by both columns — that is the grain." That produces a sparse result and picks a currency and a refund treatment by accident. The code is the easy half; the declaration is what makes it right.
They’ll ask next
The consumer needs month-over-month growth. Does your sparse result work?
UDFs vs native vs pandas UDFsRDD vs DataFrame vs Dataset
Keep only the items in each order over 100 units and sum their value — without changing the row count. What do you reach for?
Why they ask this
Higher-order functions replace an explode-filter-regroup round trip with two shuffles, and most PySpark users do not know they exist.
Say this
F.filter and F.aggregate over the array column, both of which take a lambda and run in the JVM. No explode, no regroup, no shuffle — the row count never changes.
The reasoning
F.filter(col, lambda x: ...) applies a predicate to each element and returns a narrowed array. F.transform maps over elements, F.aggregate folds them to a scalar, and F.exists and F.forall answer membership questions. All of them take Python lambdas that Spark translates into native expressions rather than Python execution — so despite the lambda, nothing leaves the JVM.
The alternative most people write is explode, filter the flattened rows, then groupBy the original key and collect_list back. That is two shuffles and a reassembly, and it risks losing rows whose array became empty. The higher-order form is a narrow transformation with none of that.
The name is the thing to know, because these are hard to find if you do not know they exist — they arrived in Spark 2.4 as SQL functions and later as PySpark wrappers, and most tutorials predate them. If a candidate reaches for explode reflexively for array work, this is the follow-up that separates the ones who have read the function list.
Two shuffles and a reassembly, and orders whose array empties out disappear.
A Python UDF over the arrayavoid
F.udf(lambda items: sum(...), 'double')
Serialises every array into Python and blocks pushdown, for logic the JVM can express.
The answer most people give
"The lambda means it runs in Python, so it is a UDF." These lambdas are used to *build* a Spark expression at plan time, not called per row. The plan shows native operators and there is no BatchEvalPython node.
They’ll ask next
You need to sort each array by a field. Is there a function for that?
RDD vs DataFrame vs DatasetJoin strategies (broadcast / SMJ / skew join / salting)
A column has nulls. Say what count(col), count(*), sum(col) and avg(col) each return, and which one silently gives a different answer from what people expect.
Why they ask this
Null semantics in aggregates are identical to SQL's and catch people constantly, and avg is the one that quietly disagrees with the intuition.
Say this
count(col) skips nulls, count(*) counts rows, sum skips nulls and returns null for an all-null group, and avg divides by the non-null count — so avg is not sum/count(*), which is where the surprise is.
The reasoning
count(col) counts non-null values; count(*) counts rows regardless. That difference is a free null check — comparing the two per group tells you the null rate without a separate pass, which is a genuinely useful trick for a data-quality model.
sum ignores nulls and returns null rather than 0 when every value in a group is null, which propagates into whatever you do next. avg is the one that surprises: it is sum over the *non-null* count, so a group of [10, null, 20] averages to 15, not 10. If the nulls mean zero, avg is wrong and you have to coalesce first — and nothing anywhere will tell you.
The related trap is join keys. A null never equals a null, so rows with null keys match nothing in an inner join and vanish. Spark's plan even makes that explicit by inserting isnotnull filters before join keys. If null keys are meaningful in your data, they need to be handled before the join — coalesced to a sentinel, or routed separately — because the join will silently drop them.
The formulations
Decide what null means, then encode itship
F.avg(F.coalesce('score', F.lit(0))) # null means zero
F.avg('score') # null means unknown
The two are different questions with different answers. Writing which one you meant is the fix.
Use count(*) - count(col) as a null-rate checkship
A free per-group null rate from the same scan, useful as a quality assertion.
Assume avg = sum / row countavoid
F.avg('score') # divides by non-null count, not by rows
Silently different whenever nulls exist, and there is no error to notice.
The answer most people give
"Spark treats nulls as zero in aggregates." It skips them, which is different — sum of an all-null group is null rather than 0, and avg divides by a smaller denominator than you expect.
They’ll ask next
Your join drops 3% of rows and you cannot see why. What do you check?
UDFs vs native vs pandas UDFsJoin strategies (broadcast / SMJ / skew join / salting)
Someone wrote a UDF that closes over a 50,000-entry Python dict to map codes to names. What is wrong with that, and what are the two better options?
Why they ask this
Closing over a large Python object is a specific and common anti-pattern with a specific cost — it is serialised with the task, not once per executor.
Say this
The dict is serialised into every task closure, so it ships hundreds of times rather than once. Better: a broadcast variable if it must stay Python, or — far better — a DataFrame joined natively.
The reasoning
A closure over a Python object is captured and shipped with each task. With 200 partitions that is 200 copies of the dictionary crossing the network and being deserialised, which dwarfs the lookup work itself and grows with parallelism rather than with data.
A broadcast variable fixes the shipping: sc.broadcast(d) sends it once per executor and the UDF reads .value. That is the right fix if the mapping genuinely has to live in Python — a call into a library, a rule engine — and it removes the per-task cost entirely.
The better answer is usually to stop using Python. Turn the mapping into a small DataFrame and broadcast join it: the optimizer sees a join, the lookup runs in the JVM, the result is pushdown-friendly, and 50,000 rows is comfortably inside the broadcast threshold. That is the shape to reach for — a lookup is a join, and expressing it as one keeps it in the engine.
The formulations
Make it a DataFrame and broadcast joinship
codes = spark.createDataFrame(d.items(), 'code string, name string')
df.join(F.broadcast(codes), 'code', 'left')
The lookup becomes a join the optimizer understands, running entirely in the JVM.
Broadcast variable, if it must stay Pythonworks
b = sc.broadcast(d)
F.udf(lambda c: b.value.get(c))
Ships once per executor instead of once per task. Still opaque to Catalyst.
Close over the dict directlyavoid
F.udf(lambda c: d.get(c)) # d captured in the closure
Serialised into every task, so the cost scales with partition count rather than with data.
The answer most people give
"It is only 50,000 entries, so it is fine." The size is not the problem — the multiplier is. Small times two hundred tasks, on every stage that uses it, is how a trivial lookup becomes the dominant cost.
They’ll ask next
The mapping changes daily. Does that change which option you pick?
Join strategies (broadcast / SMJ / skew join / salting)RDD vs DataFrame vs Dataset
Your fact table has 2,000 rows and after joining a dimension it has 2,400. Name the two possible causes and how you would tell them apart.
Why they ask this
Row count changing across a join is the most common data bug in a pipeline, and there are exactly two causes — knowing both is the answer.
Say this
Either the dimension has duplicate keys and the join fanned out, or it is an outer join adding unmatched rows. Count distinct keys in the dimension to tell them apart in one query.
The reasoning
Fan-out is the usual cause: if the dimension has more than one row per key, each fact row matches several and the count multiplies. It is invisible unless you look, and it inflates every measure downstream — which is why revenue reports are the usual place it gets discovered.
The diagnostic is one query: compare count(*) against count(distinct key) on the dimension. If they differ, the key is not unique and your join will fan out. Doing that as an assertion in the pipeline rather than as an investigation is what stops it recurring — a uniqueness test on the dimension is cheap and catches this class of bug permanently.
The other cause is join type. An outer or right join adds rows for unmatched dimension entries, which is correct when you wanted them and an inflation when you did not. And the inverse is worth naming because it is just as common: an inner join *dropping* rows, either because keys do not match or because the key is null — nulls never match, so every fact row with a null foreign key silently disappears.
The formulations
Assert dimension key uniqueness before joiningship
Keeps every fact row and makes the match rate a measurable number rather than a missing one.
Inner join and move onavoid
fact.join(dim, 'k') # rows with null or unmatched keys vanish
Silently drops rows, and the row count is the only signal — which nobody is watching.
The answer most people give
"An inner join cannot change the row count." It can move it in both directions — down when keys do not match or are null, up when the right side has duplicate keys. Both are silent.
They’ll ask next
Your fact table has null foreign keys that are meaningful. How do you join without losing them?
You know one account_id is 80% of the fact table. Write the join so it does not produce one hour-long task, and say what you would try before writing any of it.
Why they ask this
Skew handling is a coding question with a config answer first, and candidates who reach straight for salting have skipped the cheaper options.
Say this
Try AQE's skew join handling first — it splits skewed partitions automatically and costs no code. If that is not enough, broadcast the small side, and only then salt the key.
The reasoning
The first answer is configuration, not code. AQE's skew join handling detects partitions much larger than the median after a shuffle and splits them across several tasks, which solves the common case with no code change at all. Checking whether it is enabled and whether the thresholds fit your data is the cheapest thing to try.
The second is to remove the shuffle entirely. If the other side is small enough to broadcast, there is no partitioning by key and therefore no skewed partition — the skew stops mattering. That is why 'is the dimension broadcastable' is worth asking before any clever technique.
Salting is the last resort because it is real complexity: append a random salt to the skewed side's key, explode the small side across every salt value so matches still find each other, join on the composite key, then aggregate away the salt. It works and it multiplies the small side by the salt factor, changes the join key, and has to be maintained. Reaching for it before trying AQE and broadcast is the signal an interviewer is watching for.
Splits oversized partitions at runtime. No code, and it handles the common case.
Broadcast the other side if it fitsship
fact.join(F.broadcast(dim), 'account_id')
No partitioning by key means no skewed partition. The skew simply stops applying.
Salt the keyworks
fact.withColumn('sk', F.concat_ws('#','account_id',(F.rand()*8).cast('int')))
dim.withColumn('salt', F.explode(F.array(*[F.lit(i) for i in range(8)])))
Works when the others do not, and it is real complexity to write, test and maintain.
The answer most people give
"Increase shuffle partitions so the skewed key spreads out." All rows with one key hash to one partition however many partitions there are. More partitions makes every *other* partition smaller and leaves the hot one exactly as it was.
They’ll ask next
After salting, how do you get back to the correct per-account totals?
Join strategies (broadcast / SMJ / skew join / salting)RDD vs DataFrame vs Dataset
Find orders whose customer is not in the suppressed list. Write it, and say why a left join with a null check is worse than the alternative.
Why they ask this
Semi and anti joins are the right tool, are underused, and the NOT IN alternative has a null trap that is genuinely surprising.
Say this
A left anti join. A left join plus isNull works and carries every column of the right side through the shuffle for nothing — and NOT IN with a subquery returns no rows at all if the list contains a single null.
The reasoning
left_anti keeps rows from the left with no match on the right and emits only the left's columns. left_semi is the mirror — rows that do have a match, again with only the left's columns. Both stop as soon as a match is determined and neither widens the row, which is why they are cheaper than the join-and-filter form.
The left-join-then-isNull version returns the same answer and does more work: it materialises the matched columns, carries them through the shuffle, and then throws them away. It also invites a fan-out bug — if the right side has duplicate keys, the left join multiplies rows before the filter, and an anti join never does.
The NOT IN trap is worth knowing because it is counter-intuitive. In SQL semantics x NOT IN (1, 2, NULL) is not false but *unknown* for every x, because x might equal the null. So a single null in the subquery makes the whole predicate return nothing, silently. An anti join has no such behaviour, which is one more reason to prefer it.
Says exactly what you mean, emits only the left's columns, and cannot fan out.
left_semi when you want the matchesship
orders.join(active, 'customer_id', 'left_semi')
Filters by existence without widening the row or duplicating on right-side duplicates.
NOT IN against a subqueryavoid
WHERE customer_id NOT IN (SELECT customer_id FROM suppressed)
A single null in the subquery makes the predicate unknown for every row, returning nothing.
The answer most people give
"left join then filter where the right side is null is the standard way." It is a common way and it does strictly more work, and it fans out on duplicate right-side keys where an anti join does not.
They’ll ask next
suppressed has duplicate customer_ids. Which of the three approaches changes its answer?
Small files problemPartitioning, repartition vs coalesce
Write daily orders partitioned by date. What decides how many files land in each date directory, and how do you avoid producing thousands of tiny ones?
Why they ask this
Partitioned writes are where the small-files problem is created, and the arithmetic — partitions times distinct values — is not obvious.
Say this
Each Spark partition writes at most one file per date it contains, so the file count is roughly partitions times distinct dates. Repartitioning by the same column first collapses it to one file per date.
The reasoning
partitionBy controls the directory layout; it does not control file count. Every task writes its own file into every date directory it happens to hold rows for, so 200 shuffle partitions spread across 30 dates can produce up to 6,000 files for what might be a few gigabytes.
The fix is to align the in-memory partitioning with the write partitioning: repartition on the same column before writing, so all rows for a date are in one task and each date directory gets one file. If a date is too large for one file, repartition on the date plus a bucketing expression to split it a controlled number of ways rather than accepting whatever the previous stage left.
The other decision is mode, and it is a correctness one rather than a layout one. overwrite with dynamic partition overwrite replaces only the partitions the batch touched, which makes a rerun idempotent; static overwrite replaces the entire table, which is almost never what a daily job wants and is a genuinely destructive default to get wrong. append is the one that duplicates on a rerun.
The formulations
Repartition by the partition column before writingship
Replaces only the partitions in the batch, so rerunning a day is safe and does not touch the rest.
partitionBy with whatever partitioning arrivesavoid
df.write.partitionBy('order_date').mode('append')
Up to one file per task per date, and append duplicates everything on a rerun.
The answer most people give
"partitionBy controls the number of files." It controls the directory structure. File count comes from how many in-memory partitions contain rows for each value, which is why the repartition before the write is the load-bearing line.
They’ll ask next
One date is ten times the others. What does your repartition produce, and what would you do instead?
Your Airflow task writes a day of data and gets retried after a timeout. Go through append, overwrite and errorIfExists and say what each leaves behind.
Why they ask this
It is the Spark-side version of the idempotency question every data engineer is asked, and the answer is a one-line config that most people have never chosen deliberately.
Say this
append duplicates the day. Static overwrite replaces the whole table, which is worse. Dynamic partition overwrite replaces just that day's partition, which is the only one that makes a retry a non-event.
The reasoning
append is the default instinct and the one that breaks. The retry writes the same rows again into the same partition, so the day now has the rows twice and nothing errors — this is exactly the non-idempotent load that a keyed merge exists to prevent, in Spark's clothing.
overwrite has two behaviours and the distinction matters enormously. With the default static mode it drops the *entire table* and writes only what this batch contains, so a job rerun for one day destroys every other day. With partitionOverwriteMode set to dynamic, it replaces only the partitions present in the DataFrame — which is the partition-overwrite pattern, and it makes the write idempotent over its window.
errorIfExists and ignore are for one-off creation rather than for pipelines: the first turns a retry into a failure, the second turns it into a silent no-op that leaves the partial first attempt in place. On a table format like Delta or Iceberg the equivalent is a MERGE or a replaceWhere, which gives you the same idempotence plus atomicity — a failed write leaves no partial state at all.
Every retry adds the rows again, silently, and nothing in the pipeline notices.
The answer most people give
"overwrite is safe because it replaces what was there." With the default static mode it replaces the whole table, not the partition — so rerunning one day can delete the other 364.
They’ll ask next
You are on plain Parquet, not Delta. A write fails halfway. What is on disk?
Small files problemPartitioning, repartition vs coalesce
A source directory holds 40,000 files averaging 200 KB. The job spends most of its time before any real work happens. What is going on, and what do you do?
Why they ask this
It is the read-side half of the small-files problem, and the listing cost surprises people who only think about task overhead.
Say this
File listing and task overhead dominate. The driver has to enumerate 40,000 objects before planning, and then each file becomes at least one task doing 200 KB of work. Compact at the source; read with a larger split size if you cannot.
The reasoning
Two costs, and the first happens before any executor starts. The driver lists the directory to build the file index, and on object storage that is paginated API calls — tens of thousands of them, serially enough to take minutes. That is the 'nothing is happening' phase people report.
Then each file is at least one split, so you get tens of thousands of tasks each doing a trivial amount of work. Task scheduling overhead is measured in milliseconds and the work is measured in milliseconds, so the ratio is terrible and adding executors barely helps — the bottleneck is per-file, not per-byte.
The real fix is upstream: compact into files in the hundreds of megabytes, which is a scheduled job on the producing side. If you cannot change the producer, raising spark.sql.files.maxPartitionBytes and openCostInBytes lets Spark pack many small files into one task, which fixes the task overhead but not the listing. On a table format like Delta or Iceberg the listing problem disappears too, because the manifest replaces directory enumeration — which is one of the strongest practical arguments for using one.
The formulations
Compact at the sourceship
scheduled OPTIMIZE / rewrite into ~256 MB files
Fixes listing and task overhead together. The only fix that addresses the cause.
Cuts task count when you cannot change the producer. The listing cost remains.
Add executors and hopeavoid
--num-executors 100 --executor-cores 4
The bottleneck is driver-side listing and per-file overhead; more executors idle through both.
The answer most people give
"Spark handles small files fine, it just makes more tasks." More tasks is the cheap half of the problem. The expensive half is the driver enumerating 40,000 objects before a single task is scheduled.
They’ll ask next
You cannot change the producer and cannot use a table format. What is your pipeline shape?
base = spark.table("orders").filter(F.col("amount") > 100).select("region", "customer_id", "amount")
by_region = base.groupBy("region").agg(F.sum("amount").alias("total"))
out = by_region.filter(F.col("total") > 0)
The code — predict the output before reading on
base = spark.table("orders").filter(F.col("amount") > 100).select("region", "customer_id", "amount")
by_region = base.groupBy("region").agg(F.sum("amount").alias("total"))
out = by_region.filter(F.col("total") > 0)
Why they ask this
Caching is the most-misused API in Spark. The interviewer wants to hear it justified by reuse and released afterwards, not sprinkled.
Say this
Cache the branch point, only if recomputing it is genuinely expensive, and count() immediately to materialise it. MEMORY_AND_DISK is the sane default, and unpersist as soon as the last branch has run.
The reasoning
The condition for caching is reuse across *actions*. A DataFrame referenced three times is recomputed three times, because it is a plan rather than a result — so if the work behind it is expensive, caching once is a genuine saving. If it is a cheap scan and filter, caching costs memory that would have been better spent on execution.
cache() is lazy, so nothing is stored until an action runs, and only for the partitions that action touched. The idiom is to follow it immediately with count() to materialise everything, otherwise the first branch pays the full computation and the second gets a partially-cached DataFrame. MEMORY_AND_DISK is the right default level: MEMORY_ONLY silently drops partitions that do not fit and recomputes them, which is the worst of both worlds.
Then release it. Cached blocks occupy the same unified memory region execution wants, so a cache left alive after its last reader is directly causing spill somewhere else. unpersist() when the branches are done. And the strongest advice is to check whether the query can be rewritten first — the union-of-two-filters case collapses to one filter and needs no cache at all, and rewriting beats caching whenever it is possible.
What it actually returns 1 shuffle, run on Spark 4.2
One branch here, so nothing is worth caching — the plan reads the table once.
"Cache anything you use more than once." Reuse is necessary and not sufficient — caching a cheap scan spends memory that execution needed, and can make the job slower by forcing spill elsewhere.
They’ll ask next
Your cached DataFrame is bigger than the memory available. What happens with MEMORY_ONLY, and with MEMORY_AND_DISK?
Structured Streaming (watermarks, checkpoints, output modes)RDD vs DataFrame vs Dataset
Your batch job reads Parquet, aggregates and writes. What has to change to make it a Structured Streaming job, and which single line is the one that makes it correct rather than merely running?
Why they ask this
Streaming is a declared topic and the readStream/writeStream swap is the easy half. The checkpoint is the line that decides whether a restart is safe.
Say this
readStream and writeStream, an output mode, and a checkpoint location. The checkpoint is the line that matters — it holds the offsets and the aggregation state, so without it a restart reprocesses everything or loses state.
The reasoning
The API is deliberately almost identical: readStream instead of read, writeStream instead of write, and the DataFrame operations in between are unchanged. That symmetry is the point of Structured Streaming, and it is why a batch job can often be converted in three lines.
The checkpoint location is the line that carries the semantics. It stores the source offsets processed so far and the running aggregation state, so a restart resumes where it stopped. Without it there is nothing to resume from. It also cannot be shared between two queries, and it ties itself to the query's plan — meaning some code changes make an existing checkpoint incompatible, which is a real operational constraint people meet on their first deploy.
Then output mode, which is a correctness choice rather than a formatting one. append emits only rows that are final and cannot be updated, so an aggregation without a watermark cannot use it — Spark refuses, because no row is ever final. complete rewrites the whole result each trigger, which only scales for small aggregations. update emits changed rows. And a watermark is what makes append legal for a windowed aggregation, by declaring how late data may be before a window can be closed.
The formulations
readStream/writeStream with a checkpoint and a watermarkship
Resumable, and the watermark is what lets a window ever be considered final.
Trigger as a scheduled micro-batchworks
.trigger(availableNow=True) # process what is there, then stop
Streaming's offset tracking with batch's operational model. Often the right middle ground.
No checkpoint locationavoid
.writeStream.format('delta').start(path)
Nothing to resume from: a restart either reprocesses from the beginning or loses the aggregation state.
The answer most people give
"outputMode is about how the data is written." It is about which rows the query is allowed to emit. append on an aggregation without a watermark is rejected outright, because no result row can ever be declared final.
They’ll ask next
Your streaming aggregation's state grows without bound. What did you forget?
UDFs vs native vs pandas UDFsPredicate & projection pushdown
Each row has a URL and you need the path and one query parameter. Someone reached for a UDF wrapping urllib. What does the DataFrame API give you instead?
Why they ask this
String work is the second-biggest source of unnecessary UDFs after conditional logic, and the native function list is larger than most people have read.
Say this
regexp_extract, split, substring_index and parse_url cover almost all of it, and they run in the JVM and stay pushdown-friendly. The UDF version leaves the JVM per row for something the engine already does.
The reasoning
parse_url is the direct answer here — it takes a URL and a part name and returns the host, path or a named query parameter. Beyond it, regexp_extract with a capture group covers most structured strings, split gives you an array to index, and substring_index handles the delimiter-counting cases people write loops for.
The reason to prefer any of them is not only speed. A native expression stays visible to Catalyst, so a filter derived from it can still be pushed toward the scan, and the whole chain compiles into the generated code with no serialisation. The UDF version breaks that for every downstream operator, not just for itself.
The honest boundary is worth stating: some parsing genuinely needs a library — a user-agent parser, a domain-specific format with a maintained Python package. When that is the case, a pandas UDF over Arrow batches is the right shape rather than a scalar UDF, and applying it as late as possible in the plan means it sees the fewest rows. Reaching for Python first, rather than after checking the function list, is what the question is testing.
Arrow batches rather than per-row serialisation. Right when the logic really is a Python library.
Scalar UDF wrapping urllibavoid
F.udf(lambda u: urlparse(u).path)
Leaves the JVM per row and blocks pushdown for everything downstream of it.
The answer most people give
"Spark has no real string functions, so a UDF is normal." The function list is extensive — regexp_extract, regexp_replace, split, parse_url, substring_index, translate — and checking it before writing Python is the habit worth having.
They’ll ask next
Your regex is expensive and the column is mostly nulls. Where do you put the filter?
Attach each event to the price band whose from/to range contains its amount. There is no equality key. Write it, and say what the plan will do if you are careless.
Why they ask this
Range joins have no hash key, so the naive form becomes a nested loop over the cross product — and the fix is not obvious.
Say this
Without an equality predicate Spark falls back to BroadcastNestedLoopJoin or a Cartesian product. Broadcast the small band table so the nested loop is local, or manufacture an equality key by bucketing the range.
The reasoning
Hash and sort-merge joins both need an equality predicate to partition on. A pure range condition gives them nothing, so the physical plan degrades to a nested loop — BroadcastNestedLoopJoin if one side is small enough to ship, and CartesianProduct if not. Seeing either in a plan is the signal that your join lost its key.
When the band table is genuinely tiny — a few dozen price bands — BroadcastNestedLoopJoin is completely fine: the small side sits on every executor and each row scans a few dozen entries locally. The plan looks alarming and the cost is not, which is worth being able to say rather than optimising reflexively.
When neither side is small, the technique is to manufacture an equality key. Bucket the continuous value — floor(amount / 1000) — on both sides, join on the bucket *and* the range condition, and let the equality do the partitioning while the range does the filtering. Bands spanning bucket boundaries have to be replicated into every bucket they touch, which is the fiddly part and the reason to check the small side really is large before reaching for it.
CartesianProduct: every row against every band, across the network.
The answer most people give
"A join is a join — Spark will pick something sensible." It picks from what the predicate allows, and a pure range predicate allows only a nested loop. The strategy is a consequence of your join condition, not of the data size alone.
They’ll ask next
Your bands table has 2 million rows. Which option now, and what breaks first?
EvergreenPredicate & projection pushdownRDD vs DataFrame vs Dataset
You are reading a CSV where a handful of rows have an extra column and one salary field reads "NotANumber". What are your read-mode options and which do you pick?
Why they ask this
Every ingestion job meets malformed input, and the three modes encode three different policies — the question is whether you know you are choosing a policy at all.
Say this
PERMISSIVE keeps everything and parks the broken rows in a corrupt-record column, DROPMALFORMED silently discards them, FAILFAST stops the job. Read permissively with an explicit schema and route the bad rows somewhere you can see them.
The reasoning
**FAILFAST** raises on the first malformed record. Right for a contract you control and want enforced — you would rather have no output than partial output. Wrong for third-party feeds, where one bad line kills a nightly load.
**DROPMALFORMED** discards bad rows and carries on. It is the dangerous one, because the job succeeds, the row count is quietly short, and nothing anywhere records what went missing. If you use it, you have accepted silent data loss.
**PERMISSIVE** — the default — sets unparseable fields to null and, if you declare a column for it, puts the raw line in `_corrupt_record`. That is the option that lets you split the read into good rows and bad rows and send the bad ones to a quarantine table with the original text attached.
Two details that decide whether this works. The corrupt-record column must be **in the schema you supply**, named to match `columnNameOfCorruptRecord`, or Spark has nowhere to put the text. And filtering on that column can fail with "referenced columns only include the internal corrupt record column" unless the DataFrame is cached or written first — a real and confusing restriction that comes from the column being populated lazily during the scan.
The type-cast case is separate and worth calling out: `"NotANumber"` in a column typed as double does not make the *row* malformed, it makes that field null. A row where the salary silently became null looks identical to a row where salary was genuinely empty, which is why casting explicitly and checking for new nulls after the cast is a different check from the corrupt-record path.
The formulations
PERMISSIVE with a corrupt-record columnship
schema = StructType([... , StructField('_corrupt_record', StringType())])
df = spark.read.option('mode','PERMISSIVE') .option('columnNameOfCorruptRecord','_corrupt_record') .schema(schema).csv(path).cache()
bad = df.filter(col('_corrupt_record').isNotNull())
good = df.filter(col('_corrupt_record').isNull()).drop('_corrupt_record')
Nothing is lost and the bad rows are inspectable. The cache is what makes the filter legal.
Succeeds with fewer rows and no record of which. Silent data loss.
inferSchema in productionavoid
spark.read.option('inferSchema', True).csv(path)
Costs an extra pass over the data and lets the types drift between runs.
The answer most people give
"Use PERMISSIVE, it handles bad data." It *tolerates* bad data — the rows still arrive, with nulls where the parse failed. Without a corrupt-record column and a filter, you have shipped nulls downstream and called it handled.
They’ll ask next
The cast to double turned three salaries into null. How is that different from a malformed row?
EvergreenUDFs vs native vs pandas UDFsRDD vs DataFrame vs DatasetPartitioning, repartition vs coalesce
What is the difference between map and mapPartitions, and when is the second one worth the extra complexity?
Why they ask this
It is the standard test of whether someone thinks in partitions rather than rows, and the answer is the fix for a whole family of per-row setup costs.
Say this
map runs your function once per row; mapPartitions runs it once per partition and hands you an iterator. Use the second whenever there is expensive setup — a connection, a client, a loaded model — so the cost is paid per partition rather than per row.
The reasoning
`map` is the obvious one and usually the right one. Given a partition of a million rows it calls your function a million times, so anything the function constructs is constructed a million times.
`mapPartitions` calls the function once per partition, passing an iterator over that partition's rows and expecting an iterator back. That gives you a place to do setup: open one database connection, build one HTTP session, deserialise one ML model, and use it for every row in the partition. With 200 partitions and a million rows, that is 200 connections instead of 1,000,000.
The cost is that you now control the iteration, and the failure mode is materialising it. Writing `rows = list(rows)` inside the function pulls the whole partition into memory and turns a streaming operation into an OOM candidate. `yield` per row and the function stays lazy.
The other thing to say is when *not* to reach for it: in the DataFrame API, most of what people write a `map` for is expressible as a native column expression, and a native expression beats both — it stays in the JVM, keeps codegen, and does not serialise rows into Python at all. `mapPartitions` is for genuinely external work, not for arithmetic.
The formulations
mapPartitions with per-partition setupship
def go(rows):
conn = connect() # once per partition
for r in rows:
yield enrich(conn, r) # lazy — never materialised
rdd.mapPartitions(go)
One connection per partition, and the iterator stays streaming.
map with setup insideavoid
rdd.map(lambda r: enrich(connect(), r))
One connection per row. This is the bug the question is about.
mapPartitions that materialisesavoid
def go(rows):
rows = list(rows) # whole partition into memory
return [enrich(r) for r in rows]
Loses laziness and turns a large partition into an OOM.
A native expression, where it fitsship
df.withColumn('total', col('qty') * col('price'))
Neither is needed. Stays in the JVM and keeps codegen.
The answer most people give
"mapPartitions is faster than map." Not by itself — the per-row work is identical. It is faster only when there is per-call overhead to amortise. For plain arithmetic the two are the same, and the native column expression beats both.
They’ll ask next
Each partition now opens a connection and you have 5,000 partitions. What breaks?
