PySpark Interview Questions

"When I submit a PySpark job, the driver starts first. It runs my Python script, creates the SparkSession and turns my DataFrame code into a plan. The driver asks the cluster manager, which could be YARN, Kubernetes or Spark's own standalone manager, for resources, and the cluster manager launches executors on the worker nodes. Executors are JVM processes. Each one has a few cores, so it runs several tasks at once, and each task handles one partition of data. Executors also keep cached data and shuffle files. The driver tracks every task and retries the ones that fail. The PySpark twist is that my Python code drives the JVM through Py4J. Plain DataFrame operations run inside the JVM, and Python worker processes only start on the executors when I use Python UDFs or RDD lambdas."

"An action like count or a write triggers a job. Spark looks at the plan behind it and cuts it into stages wherever data has to be shuffled. Everything between two shuffles, like a read, a filter and a select, is pipelined into one stage, so rows flow through those steps without being written out. Each stage runs as a set of tasks, one per partition. So if I read a table that splits into 400 partitions, filter it and then group by country, the first stage has 400 tasks that read, filter and do a partial aggregation, then write shuffle files. The second stage reads those shuffle files and finishes the aggregation, with one task per shuffle partition. Tasks run in parallel on executor cores, which is why the number and size of partitions matter so much."

"Catalyst takes my DataFrame code as a logical plan. First it resolves column names and types against the schema. Then it applies rules to optimise it: it pushes filters down close to the data source, drops columns I never use, folds constants and merges steps. After that it picks a physical plan, for example which join strategy to use, and whole-stage code generation compiles chains of operators into efficient JVM code. When I call explain with the formatted or extended mode, I read the physical plan from the bottom up. At the bottom I check the file scan: PushedFilters and ReadSchema tell me whether my filter and column pruning reached the Parquet reader. Every Exchange is a shuffle. I look at the join node to see whether it's a BroadcastHashJoin or a SortMergeJoin, and a HashAggregate that appears twice is the partial and final aggregation around a shuffle."

"An RDD is Spark's original building block: a distributed collection of objects that I transform with functions like map and filter. Spark doesn't know what's inside those objects, so it can't optimise much. A DataFrame has a schema, named columns with types, and I describe what I want with column expressions. That lets the Catalyst optimizer reorder filters, prune columns and pick join strategies, and execution uses a compact binary format. In PySpark the gap is even bigger. With RDDs, my lambdas run in Python worker processes, so every row gets serialised from the JVM to Python and back. With DataFrames, built-in functions run inside the JVM and Python only sends the plan. The typed Dataset API exists only in Scala and Java, so in Python it's DataFrames. I'd only drop to RDDs for unusual unstructured data or old code."

"Lazy evaluation means that when I call a transformation like filter, select or join, Spark doesn't touch the data. It just adds a step to a plan, which is a DAG of operations. Only when I call an action, like count, show, collect or a write, does Spark optimise that whole plan and run it on the cluster. That's useful because Spark can see everything I asked for before it starts, so it can push filters down to the file reader, skip columns I never use and run several steps in one pass. It also explains a common surprise: a typo in a column name is usually caught straight away, when Spark analyses the plan, but a bad value that breaks a function often only fails later, at the action, because that's when the data is really read."

"Both end up with one result per key, but they shuffle very different amounts of data. With reduceByKey, Spark applies my function inside each partition before the shuffle, so if a partition has a million rows for key A, it sends one partial sum instead of a million numbers. Then it combines the partial results on the other side. With groupByKey, every value crosses the network and Spark builds a full list of values for each key on one executor. On a popular key that list can be huge and cause an out-of-memory error. So for sums, counts or anything associative I use reduceByKey or aggregateByKey. In DataFrame code I don't need to think about it as much, because groupBy with agg does that partial aggregation for me, and I can see it as two HashAggregate steps in the plan."

"In a narrow transformation, each output partition is built from exactly one input partition. Filter, select and withColumn are like that: an executor can process its partition alone, and Spark chains several of them in one stage. A wide transformation needs rows from many partitions to produce one output partition. A groupBy has to bring all rows with the same key together, and a join has to line up matching keys from both sides, unless one side is broadcast. That movement is a shuffle. Each task writes its output to local disk split by target partition, and the next stage pulls those pieces over the network. So a wide step costs disk writes, network transfer and serialisation, it's where skew shows up, and it's a stage boundary. When I tune a job, I start by counting the shuffles."

"Repartition does a full shuffle. I can go up or down in partition count, the rows end up spread fairly evenly, and I can repartition by a column so all rows for one key land together. Coalesce only reduces the number of partitions, and it does it by merging neighbouring partitions on the same executors, so there's no full shuffle, which makes it cheaper. The catch is that the resulting partitions can be uneven. There's a subtler trap too. Because coalesce doesn't create a new stage, if I write coalesce(1) at the end of a heavy job, the work before it can collapse into a single task. In that case repartition(1) is often faster, because the heavy part stays parallel and only the final write runs in one task. I use coalesce to trim partition count before a write, and repartition to fix skewed or too-few partitions."

"It sets the number of partitions Spark creates after a shuffle in DataFrame and SQL code, so it decides how many tasks a join or a groupBy stage gets. The default is 200, which is fine for small data and wrong for most big jobs. If a shuffle moves a couple of terabytes into 200 partitions, each task handles about ten gigabytes, spills to disk and may run out of memory. If a small job shuffles a few megabytes into 200 partitions, most tasks do almost nothing and the write produces lots of tiny files. My rule of thumb is to look at the shuffle write size in the Spark UI and divide it so partitions land around a hundred to two hundred megabytes. With Adaptive Query Execution on, I usually set it on the high side and let AQE merge small partitions after it sees the real sizes."

"Cache is just persist with the default storage level. For a DataFrame, that level keeps data in memory and spills to disk if it doesn't fit. Persist lets me pick the level myself, for example disk only when memory is tight. Caching is lazy, so nothing is stored until the first action runs on that DataFrame, and that first action pays the full cost. It only helps when I reuse the same result. If a cleaned table feeds three different aggregations and a write, caching it means Spark reads and cleans the source once instead of four times. If a DataFrame is used once, caching just wastes memory and can push other data out. I check the Storage tab in the Spark UI to see how much actually got cached, and I call unpersist when the reuse is over."

"The driver is one process, so anything that gathers data there can sink it. The most common cause is collect or toPandas on a big DataFrame: every row travels to the driver, and in PySpark it then becomes Python objects, which take even more memory. The second is a broadcast join where the small side isn't small, because Spark gathers the table on the driver before sending it to executors. The third is metadata: listing millions of small files, tracking hundreds of thousands of tasks, or a plan that grew huge because I built columns in a long loop. My fix depends on which it is. I write results to storage instead of collecting, aggregate first if I really need data locally, turn off or lower the broadcast threshold, and only then raise driver memory or the max result size setting."

"Inside each executor's heap, Spark keeps a small reserve, then gives a fraction of the rest, 0.6 by default, to a unified pool that execution and storage share. Execution is memory for shuffles, joins, sorts and aggregations; storage is cached data. Execution can evict cached blocks when it needs room, down to a protected storage share. The rest of the heap is user memory for my own objects. Separately, there's memory overhead outside the heap, which covers native buffers and, importantly for PySpark, the Python worker processes. So I first read the error. A Java heap out-of-memory usually means partitions are too big or skewed, so I add partitions, fix the skew, cut cores per executor so each task gets more memory, or cache less. If the cluster manager killed the container for exceeding its limit, that's usually overhead, so I raise spark.executor.memoryOverhead, especially when Python or pandas UDFs are involved."

"Lineage is Spark's record of how every partition was built, so if an executor dies it can recompute just the lost pieces. That's great until the plan gets very long. In an iterative job, say fifty rounds of joins and updates in a loop, every round adds to the plan. The optimiser then takes longer each round, I can even hit a stack overflow, and a late failure means recomputing all the way back. Cache doesn't fix that, because it keeps the data but also keeps the full lineage. Checkpoint writes the data to a reliable directory, like HDFS or object storage, and replaces the plan with a simple read of those files. So in a loop I checkpoint every few iterations. localCheckpoint is quicker because it uses executor storage, but losing an executor then loses data. Writing to Parquet and reading back is the manual version of the same idea."

"I'd start by asking which of those DataFrames are actually used more than once, because that's the only case where cache helps. In a straight pipeline that reads, transforms and writes once, every cache is pure cost. It takes executor memory that shuffles and joins need, it can push other cached data out, and the first action has to build and store it. It can even block some optimisations, because the plan reads from the cached data instead of pushing filters into the source files. So I'd suggest we keep caches only where a DataFrame feeds several outputs, add unpersist when it's done, and prove it with numbers: run it both ways and look at total time and the Storage tab. I'd frame it as a learning point, not a rejection, since caching is a common first instinct."

"In a normal join of two big tables, both sides are shuffled so matching keys land together. In a broadcast join, Spark sends a full copy of the small table to every executor, and each partition of the big table joins against it locally, so the big side never moves. That can turn a slow join into a fast one. Spark does it automatically when its size estimate for one side is under the broadcast threshold, about 10 megabytes by default, and I can force it with the broadcast function or a hint when I know the table is small after filtering. It hurts when the table isn't really small. It's collected on the driver and held in memory on every executor, so an oversized broadcast gives driver memory errors or broadcast timeouts. I also check explain to confirm I got a BroadcastHashJoin."

"First I confirm it's skew. In the stage's task summary, most tasks finish in seconds and one reads gigabytes and runs for ages. Often it's a null or default key. If it's nulls, they can't match anyway, so I split them off before the join and add them back after if it's a left join. If it's a real hot key, I salt. On the big side I add a salt column with a random number from zero to N minus one. On the smaller side I copy every row N times, once per salt value. Then I join on the key and the salt. Every big row still finds its match, because its salt value exists on the other side, but the hot key is now split across N tasks. The cost is that the small side grows N times, so I keep N modest. Before salting by hand, I check whether a broadcast join works or whether AQE's skew join handling already splits it."

"For an equality join, Spark has three main options. Broadcast hash join sends the small side to every executor and is the fastest when one side fits comfortably. Sort-merge join shuffles both sides by the join key, sorts each partition and merges them, and it's the default for two large tables because it can spill to disk and still finish. Shuffle hash join also shuffles both sides but builds a hash table per partition instead of sorting, which can win when one side is much smaller but too big to broadcast. For joins with no equality condition, like a range or inequality match, Spark falls back to a broadcast nested loop join or a cartesian product, which is where jobs blow up. I check the choice with explain, and I steer it with hints like broadcast, merge or shuffle_hash, or by adding an equality key so a range join isn't a nested loop."

"A regular Python UDF runs outside the JVM. Spark has to serialise each row, send it to a Python worker process, run my function, and send the result back. That's slow on its own, and the optimiser treats the UDF as a black box, so it can't push filters through it or generate efficient code around it. So my first move is always to rewrite the logic with built-in functions: string cleanup, dates, conditions with when and otherwise, regex, arrays and maps are almost all covered, and they run in the JVM. If I really need Python logic, like a library call, I use a pandas UDF. It sends data in Arrow batches and my function works on a whole pandas Series at a time, which is much faster than row by row, though still slower than a built-in. And I remember a UDF has to handle nulls itself."

"I'd use a window. I partition by customer_id so each customer is handled on their own, and order by updated_at descending so the newest row comes first. Then row_number gives each row its position, and I keep only row one and drop the helper column. Two details matter. First, if two updates share the same timestamp, row_number picks one of them in a way I shouldn't rely on, so I add a tie-breaker like an ingestion id to the ordering. If I wanted to keep all tied rows I'd use rank instead. Second, I wouldn't use dropDuplicates on customer_id, because it keeps whichever row it happens to meet first, which isn't reliably the latest one. The window needs a shuffle by customer, which is fine, but a window with no partitionBy would pull everything into one partition."

"I define one window partitioned by user_id and ordered by order_date. For the running total I sum amount over that window, but I set the frame explicitly with rowsBetween, from unbounded preceding to the current row. If I leave the default, Spark uses a range frame when there's an ordering, so two orders on the same date would both show the total including each other, which surprises people. For the gap, lag gives me the previous row's order date within the same user, and datediff turns the two dates into a number of days. The first order for each user has no previous row, so lag returns null and the gap is null, which I'd keep rather than fake as zero. If orders can share a date, I'd also add an order id to the ordering so the result is stable."

"I group by region and use agg so I can compute several things in one pass: the sum of amount as revenue, and a count of distinct order ids as orders, in case an order has several line rows. I alias each one so the columns have clean names. Then I filter on revenue, which works like a HAVING clause in SQL because it runs on the aggregated result, and order by revenue descending. If there were row-level conditions, like only completed sales, I'd put that filter before the groupBy, because then fewer rows get shuffled. One thing I'd mention on big data is that an exact distinct count is expensive. If an estimate is fine for a dashboard, approx_count_distinct is much cheaper."

"Parquet stores data by column, not by row. So a query that needs three columns out of fifty only reads those three, and similar values sit together, which compresses well. It also stores the schema and min and max statistics for chunks of rows, so Spark can push a filter down and skip chunks that can't match. partitionBy is about folders. When I write with partitionBy on event_date, Spark creates one directory per date and doesn't store that column inside the files. When someone later filters on event_date, Spark only lists and reads the matching folders, which is partition pruning. The choice of column matters. A date or country that queries filter on is good. A high-cardinality column like user_id is terrible, because it creates millions of folders full of tiny files."

"Small files hurt because the overhead is per file. Spark has to list them, open each one and read its footer, and on object storage every one of those is a network call, so a job can spend longer on metadata than on data. On HDFS they also load up the NameNode. They usually come from how we write. If a stage has 200 tasks and the data spans 100 dates with partitionBy, each task can write a file into every date folder, so I get up to twenty thousand files. Streaming jobs and frequent small appends do the same over time. To fix an existing table, I run a compaction job that reads each partition and rewrites it as a few large files. To prevent it, I repartition by the partition column before the write, so each date is written by a single task, and cap giant files with maxRecordsPerFile."

"With inferSchema on a CSV, Spark scans the data once just to guess the types, then reads it again for real, which is wasted time on large inputs. The bigger problem is that guesses can be wrong or change. A postal code column becomes an integer and loses its leading zeros. A column that's all empty today is typed as string, and tomorrow it's different, so a downstream job breaks. With an explicit schema, the read is faster and the types are a contract I control. I also set the read mode on purpose. PERMISSIVE, the default, turns bad values into nulls, and I can capture the raw bad line in a corrupt record column. FAILFAST stops the job on the first bad row, which I prefer for inputs that should never be malformed. I can write the schema as StructType or a short DDL string."

"Append is the cause: a rerun writes the same rows a second time, and a half-failed run can leave partial files too. I'd change the job so each run owns its partitions. I set the partition overwrite mode to dynamic and write with mode overwrite, so Spark replaces only the date folders present in today's output and leaves every other day alone. Then running it once or five times gives the same result. I'd be careful here, because in the default static mode, overwrite with partitionBy deletes the entire table, which is a far worse incident than duplicates. For the damage already done, I'd rebuild the affected days from the source rather than trying to delete duplicates in place. If the team needs atomic replaces or row-level merges, a transactional table format like Iceberg or Delta Lake is the longer-term fix."

"Without AQE, Spark picks the whole physical plan up front from estimates, which are often wrong after filters and joins. With AQE, which is on by default in recent Spark 3 versions, Spark pauses at each shuffle boundary, looks at the real size of every shuffle partition, and re-plans the rest. It does three main things for me. It merges small shuffle partitions, so I don't get thousands of tiny tasks. It can switch a sort-merge join to a broadcast join when one side turns out to be small after filtering. And it can split a skewed partition in a sort-merge join into several tasks. It has limits, though. It only acts at shuffle boundaries, and its skew handling is aimed at joins, so a skewed groupBy still needs me. In explain, the plan shows as AdaptiveSparkPlan, and the final plan is only known once the query has run."

"No code change means the data, the plan or the cluster changed, so I compare the slow run with a good one in the Spark UI or history server. First I find which stage got slower, because it's rarely all of them. Inside that stage I look at the task summary. If the max task takes an hour and the median takes a minute, it's skew, maybe a new hot key. If all tasks are slower and there's lots of spill, the data grew and partitions are too big. If the task count exploded, I check for small files upstream. Then I open the SQL tab and compare plans, because a lookup table that grew past the broadcast threshold turns a fast broadcast join into a full shuffle join. Last I check the Executors tab for lost executors, high GC time, or fewer executors than usual because the cluster was busy."

"At my last company we had a nightly job joining a day of clickstream events to a product table, and it had crept up to about two hours, which pushed our morning reports late. In the Spark UI one stage took almost all the time, and in its task summary the median task finished in under a minute while one task ran for over forty. That's skew. Checking the key counts, a big chunk of events had a null product id from an old tracking bug. Nulls can't match in the join anyway, so I split them off before the join and unioned them back after. I also noticed the product table was small once I selected only the columns we used, so I broadcast it and the big shuffle went away. The job dropped to under half an hour, and I started checking task-time spread on every new job."

"In my previous team, a scoring job started failing after the input doubled. Executors kept getting lost and the retries failed too. The first thing I did was read the executor logs rather than just the driver's stack trace, and they showed the cluster manager was killing containers for exceeding their memory limit. It wasn't a Java heap error. That pointed outside the heap. The job used a pandas UDF to run a model, so each Python worker was holding big Arrow batches plus the model in memory, and that all counts as overhead. I raised the executor memory overhead, lowered the Arrow batch size so each Python call held fewer rows, and reduced cores per executor so fewer Python workers shared a container. It ran cleanly after that. I also added a note to our runbook on telling heap errors from container kills."

"At my last company I moved a pandas script that built daily user features, because the input had grown past what one machine could hold in memory. The biggest change was in how I thought, not the syntax. The pandas version looped over rows and used apply with Python functions everywhere, and a straight translation into UDFs was slow. I rewrote those parts with built-in column functions and window functions. I also had to drop assumptions about row order, because a DataFrame in Spark has no index and no guaranteed order unless I sort. And I learned to keep everything lazy and only write at the end, instead of calling collect to peek at results mid-script. To prove it was right, I ran both versions on a sample and compared the outputs row by row before switching over."