Interview Prep Guide
PySpark Data Engineer Interview Questions and Production Scenarios
Practical PySpark questions covering DataFrames, transformations, joins, partitioning, skew, UDFs, testing, optimization, and production troubleshooting.
Applied PySpark Questions
How are PySpark DataFrames different from RDDs?
DataFrames provide a schema and declarative operations that Spark can optimize, while RDDs expose lower-level distributed objects with fewer automatic optimizations.
What are transformations and actions in PySpark?
Transformations build a lazy logical plan; actions trigger execution and return or write a result.
Why should you prefer built-in Spark SQL functions over Python UDFs?
Built-in functions remain visible to Spark optimization and avoid much of the Python serialization boundary, while Python UDFs can become opaque and expensive.
How would you test a PySpark transformation?
Use small deterministic DataFrames with explicit schemas, assert content without relying on row order, and cover nulls, duplicates, boundaries, and invalid input.
Advanced PySpark Questions
What is the difference between repartition and coalesce?
Repartition performs a shuffle to redistribute data and can increase or decrease partitions; coalesce usually reduces partitions with less movement but may create imbalance.
How do you choose between broadcast and sort-merge joins?
Broadcast a genuinely small side to avoid a large shuffle; use sort-merge for large compatible inputs, supported by good partitioning, statistics, and adaptive execution.
How do you handle skewed keys in a PySpark job?
Confirm skew from task and key-distribution evidence, then use techniques such as pre-aggregation, salting, selective broadcast, skew hints, or adaptive query execution.
When should you cache, persist, or checkpoint a DataFrame?
Persist only reused expensive results when recomputation costs more than storage; checkpoint when lineage truncation and recovery boundaries matter.
PySpark Production Scenarios
A PySpark job became three times slower after a data-volume increase. How do you investigate?
Compare recent runs in the Spark UI, inspect stage and task time, shuffle, spills, skew, input files, plan changes, and cluster saturation before tuning.
A daily pipeline creates thousands of tiny output files. How would you fix it?
Control output partition count and partition columns near the write, compact existing data safely, and avoid repartitioning by highly granular or high-cardinality values.
Executors repeatedly fail with out-of-memory errors during a join. What do you do?
Identify whether the pressure comes from broadcast size, skew, shuffle blocks, wide rows, caching, or excessive per-executor concurrency, then correct that cause.
How would you make an incremental PySpark pipeline idempotent?
Use a stable source position or business key, deterministic transformations, transactional merge or partition replacement, deduplication rules, and recorded checkpoints.
PySpark DataFrame and Pipeline Depth
What is the difference between narrow and wide transformations in PySpark?
Narrow transformations read each output partition from a limited set of input partitions; wide transformations redistribute data across partitions and usually create a shuffle boundary.
When should you use broadcast joins in PySpark?
Broadcast the genuinely small side of a join when it safely fits executor memory and avoids a large shuffle, validating the decision with statistics and the physical plan.
How do repartition and coalesce differ?
Repartition performs a shuffle to increase or decrease partitions with better redistribution; coalesce usually reduces partitions with less movement but can preserve imbalance.
When should a PySpark DataFrame be cached?
Cache only reused, expensive-to-recompute data when the saved work exceeds memory, serialization, and eviction costs, then unpersist it when no longer needed.
Why should production PySpark pipelines use explicit schemas?
Explicit schemas prevent repeated inference, document the contract, preserve intended types, and allow malformed or evolving input to fail or quarantine predictably.
How do null values affect PySpark joins and filters?
SQL null uses three-valued logic, so ordinary equality does not match null to null and filters can discard unknown predicates unless handled explicitly.
How do you prevent the small-files problem in PySpark output?
Control upstream partitioning, compact incrementally, choose sensible partition columns and file sizes, and avoid creating many tiny batches or high-cardinality directories.
How would you unit-test a PySpark transformation?
Build small deterministic DataFrames with explicit schemas, exercise edge cases, compare rows and schemas, isolate pure transformation logic, and add integration tests for storage boundaries.
PySpark Coding Round
Find every employee with the second-highest salary in each department using PySpark
Use dense_rank so employees tied on the second distinct salary are all returned. from pyspark.sql import functions as F from pyspark.sql.window import Window w = Window.partitionBy("department").orderBy(F.col("salary").desc()) result = (employees .withColumn("salary_rank", F.dense_rank().over(w)) .filter(F.col("salary_rank") == 2) .select("employee_id", "department", "salary")) A department with only one distinct salary has no rank two and naturally produces no rows.
Return the top N rows per group in PySpark with an explicit tie policy
For exactly N rows per department, use row_number with a stable tie-breaker. Use dense_rank when all ties at the Nth value must be included. from pyspark.sql import functions as F from pyspark.sql.window import Window n = 3 w = Window.partitionBy("department").orderBy( F.col("salary").desc(), F.col("employee_id").asc() ) result = (employees .withColumn("row_num", F.row_number().over(w)) .filter(F.col("row_num") <= n) .drop("row_num"))
Calculate a customer running total in PySpark
Define the cumulative frame explicitly and include an order ID tie-breaker. from pyspark.sql import functions as F from pyspark.sql.window import Window w = (Window.partitionBy("customer_id") .orderBy("order_date", "order_id") .rowsBetween(Window.unboundedPreceding, Window.currentRow)) result = orders.withColumn("running_total", F.sum("amount").over(w)) Decide whether NULL amounts should be ignored, treated as zero, or rejected during validation.
Calculate a three-record moving average in PySpark
Average the current employee with the two previous hires within each department. from pyspark.sql import functions as F from pyspark.sql.window import Window w = (Window.partitionBy("department") .orderBy("hire_date", "employee_id") .rowsBetween(-2, 0)) result = employees.withColumn("moving_avg_salary", F.avg("salary").over(w))
Compare each period with its previous value using lag in PySpark
Partition when the dataset contains several products or accounts; otherwise values can leak across entities. from pyspark.sql import functions as F from pyspark.sql.window import Window w = Window.partitionBy("product_id").orderBy("period") result = (sales .withColumn("previous_sales", F.lag("sales").over(w)) .withColumn("sales_change", F.col("sales") - F.col("previous_sales")))
Deduplicate a PySpark DataFrame while keeping the latest record
Rank versions within the duplicate key and retain one deterministic winner. from pyspark.sql import functions as F from pyspark.sql.window import Window w = Window.partitionBy("customer_id", "product_id").orderBy( F.col("updated_at").desc(), F.col("event_id").desc() ) result = (events .withColumn("row_num", F.row_number().over(w)) .filter(F.col("row_num") == 1) .drop("row_num"))
Calculate and rank product revenue across two PySpark DataFrames
Join at product grain, calculate line revenue, aggregate, and then rank the result. from pyspark.sql import functions as F revenue = (orders .join(F.broadcast(products.select("product_id", "product_name", "price")), "product_id") .withColumn("line_revenue", F.col("quantity") * F.col("price")) .groupBy("product_id", "product_name") .agg(F.sum("line_revenue").alias("total_revenue")) .orderBy(F.col("total_revenue").desc())) top_three = revenue.limit(3) Broadcast only when the product dimension is safely small enough for every executor.
Join orders to customers without inflating customer totals
Validate that the customer dimension has one row per key before the join; a duplicated dimension silently multiplies totals. from pyspark.sql import functions as F customer_dim = customers.dropDuplicates(["customer_id"]) result = (orders .join(customer_dim.select("customer_id", "customer_name"), "customer_id", "left") .groupBy("customer_id", "customer_name") .agg( F.sum("amount").alias("total_spent"), F.avg("amount").alias("average_order_value"), F.countDistinct("order_id").alias("order_count") )) In production, reject unexpected duplicate dimension keys rather than silently choosing an arbitrary row.
Calculate month-over-month revenue change in PySpark
Aggregate raw events first, then apply lag at monthly grain. from pyspark.sql import functions as F from pyspark.sql.window import Window monthly = (sales .withColumn("month", F.trunc("sale_date", "month")) .groupBy("product_id", "month") .agg(F.sum("revenue").alias("revenue"))) w = Window.partitionBy("product_id").orderBy("month") result = (monthly .withColumn("previous_revenue", F.lag("revenue").over(w)) .withColumn("revenue_change", F.col("revenue") - F.col("previous_revenue")))
Compute a rolling three-month activity sum in PySpark
For an additive metric, aggregate to month first and apply a three-row window. Join to a calendar DataFrame first if missing months must count as zero. from pyspark.sql import functions as F from pyspark.sql.window import Window monthly = (activity .withColumn("month", F.trunc("activity_date", "month")) .groupBy("product_id", "month") .agg(F.sum("activity_count").alias("monthly_activity"))) w = (Window.partitionBy("product_id") .orderBy("month") .rowsBetween(-2, 0)) result = monthly.withColumn("rolling_3m_activity", F.sum("monthly_activity").over(w))
Show row_number, rank, and dense_rank side by side in PySpark
Use the same ordering window so the difference is visible when salaries tie. from pyspark.sql import functions as F from pyspark.sql.window import Window w = Window.partitionBy("department").orderBy(F.col("salary").desc()) result = (employees .withColumn("row_number", F.row_number().over(w)) .withColumn("rank", F.rank().over(w)) .withColumn("dense_rank", F.dense_rank().over(w))) row_number is unique, rank leaves gaps after ties, and dense_rank does not.
Pivot student subject scores into columns with PySpark
Provide the expected subject list when it is known; this avoids an eager distinct scan to discover pivot values. subjects = ["math", "science", "english"] result = (marks .groupBy("student_id", "student_name") .pivot("subject", subjects) .agg(F.max("score"))) If more than one score exists per student and subject, replace max with the business-approved resolution rule.
Flatten an array of order items with explode in PySpark
Use explode_outer when orders with null or empty arrays must remain visible. from pyspark.sql import functions as F result = (orders .withColumn("item", F.explode_outer("items")) .select( "order_id", F.col("item.product_id").alias("product_id"), F.col("item.quantity").alias("quantity"), F.col("item.price").alias("price") ))
Find customers with no orders using a PySpark anti join
Project only the distinct join key from orders so Spark moves less data. ordered_customers = orders.select("customer_id").where("customer_id IS NOT NULL").distinct() result = customers.join( ordered_customers, on="customer_id", how="left_anti" ) A left anti join returns left-side rows for which no matching right-side key exists.
Find customers purchasing in at least three consecutive months with PySpark
Convert each month to a numeric index and subtract its row number; consecutive indices then share a group key. from pyspark.sql import functions as F from pyspark.sql.window import Window months = (orders .select("customer_id", F.trunc("order_date", "month").alias("month")) .distinct() .withColumn("month_num", F.year("month") * 12 + F.month("month"))) w = Window.partitionBy("customer_id").orderBy("month_num") result = (months .withColumn("rn", F.row_number().over(w)) .withColumn("streak_id", F.col("month_num") - F.col("rn")) .groupBy("customer_id", "streak_id") .agg( F.min("month").alias("streak_start"), F.max("month").alias("streak_end"), F.count("*").alias("consecutive_months") ) .filter(F.col("consecutive_months") >= 3))