PySpark Coding Questions cheat sheet
Live-coding and whiteboard PySpark questions on transformations, joins, shuffles, skew, windows, UDFs, and Structured Streaming.
Execution model fundamentals
What is the difference between a transformation and an action?- Transformations such as select, filter, and join build the logical plan lazily and return a new DataFrame; actions such as count, collect, show, and write trigger execution. Nothing runs until an action fires, which is why a syntactically wrong column reference can surface far from the line that wrote it. The practical follow-up is that calling an action twice recomputes the whole lineage unless you cache or checkpoint.
Explain narrow versus wide transformations.- A narrow transformation lets each output partition depend on exactly one input partition — map, filter, union — so it pipelines inside a stage with no network movement. A wide transformation requires data from many input partitions, which forces a shuffle and a stage boundary. Every groupBy, join on a non-co-partitioned key, distinct, and repartition is wide, and the number of shuffles is the first thing to count when tuning a job.
How do jobs, stages, and tasks relate?- One action creates one job; the job is split into stages at shuffle boundaries; each stage runs one task per partition. That mapping is how you read the Spark UI — a stage with 200 tasks where one runs ten times longer than the median is skew, and a job with many small stages usually means excessive shuffling. Default shuffle partitions is 200 unless adaptive execution coalesces it.
What does the Catalyst optimizer actually do?- It rewrites the logical plan with rules such as predicate pushdown, projection pruning, constant folding, and null propagation, then picks physical operators using cost-based decisions where statistics exist, and finally generates JVM bytecode with whole-stage code generation. The consequence for you is that filtering after a select is normally as fast as filtering before, but anything Catalyst cannot see through — a Python UDF, for instance — becomes an optimization barrier.
When does cache actually help, and what are the pitfalls?- Cache helps when a DataFrame is used by two or more actions and recomputation is more expensive than the memory it occupies. Pitfalls are that cache is lazy so nothing is stored until the next action, that MEMORY_AND_DISK spills rather than fails but silently slows the job, and that caching a DataFrame larger than available memory evicts other work. Always unpersist when done, and prefer checkpointing when the lineage is long enough to be a recovery risk.
What is the difference between repartition and coalesce?- repartition performs a full shuffle and can increase or decrease partitions with even distribution, optionally on a key. coalesce merges partitions without a full shuffle, so it only decreases, and it can starve parallelism upstream because the reduced partition count propagates backwards through the narrow stage. Use coalesce before a write to reduce file count, and repartition when you need balance or a key-based layout.
Joins and shuffles
How do you force a broadcast join, and when should you not?- from pyspark.sql.functions import broadcast result = large_df.join(broadcast(small_df), on='dim_id', how='left') The broadcast hint ships the small side to every executor and eliminates the shuffle of the large side. Default automatic threshold is 10 MB of estimated size, raised via spark.sql.autoBroadcastJoinThreshold. Do not force it when the small side is not actually small after filtering, because collecting it to the driver first can cause an out-of-memory error on the driver, and note that it needs to fit in each executor as well.
A join on user_id is taking hours because one user has 40 percent of the rows. Fix it.- from pyspark.sql import functions as F SALT = 16 left = df.withColumn('salt', (F.rand() * SALT).cast('int')) right = dim.withColumn('salt', F.explode(F.array([F.lit(i) for i in range(SALT)]))) out = left.join(right, on=['user_id', 'salt'], how='inner').drop('salt') Salting splits the hot key across sixteen reducers by replicating the small side sixteen times. First try adaptive query execution skew join handling, which splits large partitions automatically when spark.sql.adaptive.skewJoin.enabled is on, and only hand-salt when that is insufficient or you are on a fixed plan.
What join strategies does Spark have and how does it choose?- Broadcast hash join when one side is under the threshold, sort-merge join as the default for large-to-large equi-joins, shuffle hash join when one side is much smaller but still too big to broadcast and sorting is not worth it, and broadcast nested loop join as the fallback for non-equi and cross joins. Non-equi joins are the trap — a range condition with no equality predicate collapses to nested loops and can effectively never finish.
Why did your left join return more rows than the left table had?- The right side is not unique on the join key, so each duplicate multiplies the matching left row. Verify with a groupBy on the key and a having count greater than one, then decide whether to deduplicate the right side with a ROW_NUMBER-style window or whether the fan-out is legitimate and the aggregation downstream needs to change. This is the single most common correctness bug in interview code and the interviewer is watching whether you check.
How do you avoid a shuffle on repeated joins over the same key?- Bucket the tables on the join key with the same bucket count when writing, so Spark can perform a bucketed sort-merge join without shuffling either side. The constraints are that both tables must use the same number of buckets and the same key, and that bucketing only pays off if the tables are joined repeatedly. Alternatively, on a lakehouse, sorting and z-ordering the storage helps pruning but does not eliminate the shuffle.
Aggregations and window functions
Deduplicate a DataFrame keeping the most recent row per key.- from pyspark.sql import Window from pyspark.sql import functions as F w = Window.partitionBy('user_id').orderBy(F.col('updated_at').desc(), F.col('event_id').desc()) latest = (df.withColumn('rn', F.row_number().over(w)) .filter(F.col('rn') == 1) .drop('rn')) Add the tiebreaker on event_id so the result is deterministic when two rows share a timestamp. dropDuplicates(['user_id']) is shorter but picks an arbitrary row, which makes reruns nondeterministic and tests flaky.
What is the default window frame and why does it matter?- With an orderBy and no explicit frame, the default is RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW; without an orderBy it is the whole partition. That default makes last_value return the current row rather than the partition maximum, so you must specify Window.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing). The rowsBetween versus rangeBetween distinction is a standard follow-up — rows counts physical rows, range compares values, so ties are grouped.
Compute a 7-day rolling sum per user on daily data with gaps.- from pyspark.sql import Window from pyspark.sql import functions as F days = lambda n: n * 86400 w = (Window.partitionBy('user_id') .orderBy(F.col('event_date').cast('timestamp').cast('long')) .rangeBetween(-days(6), 0)) out = df.withColumn('rolling_7d', F.sum('amount').over(w)) rangeBetween over a numeric timestamp gives a true calendar window that stays correct when days are missing; rowsBetween(-6, 0) would count the previous six present rows, which spans more than a week whenever there are gaps.
Why is a window function without partitionBy dangerous?- It moves the entire dataset into one partition on a single executor, so the job either takes forever or fails with an out-of-memory error, and Spark logs a warning saying exactly that. If you genuinely need a global ordering or a global row number, use zipWithIndex on the RDD, or compute per-partition counts and offset them, or reconsider whether you need it at all.
How do you pivot in PySpark and what is the performance catch?- out = (df.groupBy('store_id') .pivot('month', ['2026-01', '2026-02', '2026-03']) .agg(F.sum('revenue'))) Passing the explicit list of pivot values avoids an extra job that Spark otherwise runs to discover distinct values, which on a large table is a full scan. Pivoting on a high-cardinality column produces a very wide DataFrame and is almost always the wrong shape for downstream storage.
UDFs and Python interop
Why are Python UDFs slow and what do you use instead?- A plain Python UDF serializes every row to a Python worker process and back, breaking whole-stage code generation and blocking Catalyst from pushing predicates through it. Prefer built-in functions in pyspark.sql.functions, which run entirely in the JVM. When you truly need Python, use a pandas UDF, which uses Apache Arrow to transfer columnar batches and typically runs several times faster.
Write a pandas UDF and explain when it wins.- import pandas as pd from pyspark.sql.functions import pandas_udf @pandas_udf('double') def zscore(v: pd.Series) -> pd.Series: return (v - v.mean()) / v.std() out = df.withColumn('z', zscore('amount')) Arrow moves data as columnar batches so serialization cost is amortized across the batch and the function body runs as vectorized pandas. It wins for numeric and vectorizable logic; it does not rescue genuinely row-by-row Python or calls to a slow external service.
How do you call an external API for every row without destroying the cluster?- Use mapInPandas or a grouped map so you process a batch per call with connection reuse and a bounded thread pool, add retries with jitter, and repartition to control concurrency — partition count is your rate limit. Better still, avoid it in the job entirely and stage the lookups into a table you can join. Interviewers ask this to see whether you recognize that one call per row times ten million rows is not a design.
What does the driver do, and what puts it at risk?- The driver builds the plan, schedules tasks, and receives results from actions. It falls over when you call collect or toPandas on a large DataFrame, broadcast a table that is bigger than driver memory, or accumulate a huge lineage or too many small tasks. Use limit before collect, write to storage instead of collecting, and check spark.driver.maxResultSize, which defaults to 1 GB and will abort the job rather than silently hang.
How do you unit test PySpark code?- Structure transformations as pure functions taking and returning DataFrames so they can be called with small in-memory fixtures created by spark.createDataFrame. Share one local SparkSession across the test session because startup dominates runtime, and compare results with the built-in assertDataFrameEqual, or by sorting and collecting for older versions. Do not test against production tables — deterministic fixtures are the point.
Performance tuning
What is Adaptive Query Execution and what does it change?- AQE, on by default since Spark 3.2, re-optimizes the plan at runtime using actual shuffle statistics. It coalesces small shuffle partitions so you no longer hand-tune the 200 default, converts sort-merge joins to broadcast joins when the measured side turns out to be small, and splits skewed partitions. The follow-up is when it does not help — it cannot fix skew inside a single unsplittable key group and it needs a shuffle to have statistics to work with.
How do you diagnose a slow stage from the Spark UI?- Look at the stage task summary and compare max to median duration and shuffle read size — a large gap is skew. Check spill to memory and disk, which means partitions are too big for the executor. Check the number of tasks against the number of cores, since far fewer tasks than cores wastes the cluster and far more adds scheduling overhead. Then look at the SQL tab plan to confirm the join strategy and whether scans pruned partitions.
How many partitions should your data have?- Target roughly 128 MB to 200 MB per partition and at least two to three tasks per core so stragglers can be balanced. spark.sql.shuffle.partitions defaults to 200, which is wildly wrong for both tiny and very large jobs, though AQE coalescing softens the small case. State the reasoning — partition size and core count — rather than a magic number.
What causes an executor out-of-memory error and how do you fix it?- Usually a single partition that is too large from skew or explode, a broadcast that exceeded expectations, or a groupBy collecting large collections per key with collect_list. Fix by increasing partition count to shrink each partition, salting the skewed key, replacing collect_list aggregations with a windowed or joined approach, and only then by adding executor memory. Raising memory first hides the design problem and costs money forever.
Why did your job produce 50,000 tiny output files?- The write inherited the partition count of the final stage, so 200 shuffle partitions times 250 date partitions gives 50,000 files. Fix by repartitioning on the write partition columns before saving so each output partition is written by one task, or use coalesce when the total is small. Tiny files then cost you on every downstream read through listing and open overhead.
Structured Streaming
What is a watermark and what happens without one?- agg = (events .withWatermark('event_time', '15 minutes') .groupBy(F.window('event_time', '5 minutes'), 'device_id') .agg(F.count('*').alias('n'))) The watermark tells Spark the maximum lateness it will tolerate, letting it finalize and drop old window state. Without it, state for stateful aggregations grows without bound until the executor dies. Records arriving later than the watermark are silently dropped, so the tradeoff is state size against completeness, and you should monitor the dropped-late-rows metric.
Explain the output modes.- Append emits only rows that are final and will not change, which for windowed aggregations means after the watermark passes. Update emits rows whose value changed since the last trigger. Complete rewrites the entire result table every trigger and is only viable for small aggregations. Append with a watermark is the standard for writing to files or a table, because files cannot be updated in place.
How does checkpointing give you fault tolerance?- The checkpoint location stores offsets processed, a write-ahead log of the commit, and the aggregation state store, so a restarted query resumes exactly where it stopped. It must live on durable shared storage, and it is tied to the query — changing the aggregation logic or key columns can make the existing state incompatible and require a fresh checkpoint. Deleting a checkpoint to fix an error reprocesses or skips data, so treat it as a deliberate operation.
What is the difference between the availableNow and continuous triggers?- Trigger.AvailableNow processes all available data in multiple micro-batches and then stops, which makes it the right choice for a scheduled incremental batch job over a streaming source — it inherits checkpointing without an always-on cluster. Continuous processing is the experimental low-latency mode with at-least-once semantics and limited operator support; in practice micro-batch with a short interval is what production uses.
How do you join a stream to a slowly changing dimension?- A stream-to-static join re-reads the static side on each micro-batch, so a Delta or Iceberg table picks up updates without restarting the query, and there is no state to bound. A stream-to-stream join requires watermarks on both sides plus a time range condition so Spark can expire buffered rows. Choose the static form whenever the dimension can be a table, because the state management burden is the whole difficulty.
From DataLane — tutorials at/blog, practice SQL live in theplayground.