944 Spark interview questions from reported data engineer loops: Spark SQL transforms plus 12 production incidents to diagnose from real Spark UI evidence. The 20 below are solved in full, 1 per company across Databricks, Uber, Apple, Microsoft, Meta, Google, Amazon, TikTok, and 12 more.
Spark rounds are diagnosis rounds. The interviewer is rarely checking whether you can write a groupBy; they are checking whether you can read a task table where one task has run for 3 days, name the hot key, and fix it without breaking the downstream contract. The other half is the Spark SQL round, where the transform patterns of any data engineer loop reappear in Spark dialect. Between them there are 944 Spark questions to work through here, every one runnable against a real schema with a verdict at the end. The 20 solved below come from reported data engineer loops, 1 per company across Databricks, Uber, Apple, Microsoft, Meta, Google, Amazon, TikTok, and 12 more, ordered easy to hard, followed by 12 production incidents with the real Spark UI evidence attached.
Work them here, or open each incident's full workspace for the evidence tabs. When you finish these, the Spark SQL questions and Spark optimization questions continue the same round.
Computed from the 15 spark-native scenarios tagged by concept on this site, across the 9 employers they were reported from. Percentages are the share of those problems using each concept, so they overlap: a single question usually pulls in 3 or 4 at once.
| Concept | Share of problems | Employers | What comes up |
|---|---|---|---|
| Spark execution model | 33% (5) | 3 | Jobs, stages, tasks, and what actually triggers a shuffle. |
| Broadcast joins | 27% (4) | 5 | Replacing a shuffle join with a broadcast when one side fits in memory. |
| Memory management and OOM | 27% (4) | 3 | Executor memory, spill, and why a job dies at the last stage. |
| Data skew | 20% (3) | 3 | Detecting and mitigating skew with salting when one key dominates a distributed JOIN. |
| Shuffle optimization | 20% (3) | 3 | Partition counts and wide vs narrow transformations. |
| Catalyst optimizer | 20% (3) | 2 | How the logical plan is rewritten, and why UDFs defeat it. |
| Partitioning strategy | 20% (3) | 2 | Choosing a partition column that avoids small files and skew. |
| Predicate pushdown and partition pruning | 13% (2) | 2 | Filtering at the file scan so Parquet partitions are never read. |
| Reading the Spark UI | 13% (2) | 3 | Diagnosing a slow stage from the UI rather than guessing. |
These counts come from the same practice problems you can solve here, and they update as new questions come in.
7 patterns cover the Spark SQL round; the incident section below covers the diagnosis round. Each maps to questions on this page.
| Pattern | The canonical form | On this page |
|---|---|---|
| Ranking with ties | RANK or DENSE_RANK, never LIMIT, when ties must survive | Questions 1, 3, 4, 14, 19 |
| Dedup-latest and first-touch | ROW_NUMBER per key with a deliberate tiebreaker | Questions 17, 18 |
| Windows and frames | Explicit ROWS BETWEEN; running totals with composite ORDER BY | Questions 7, 15 |
| Conditional aggregation and pivots | SUM(CASE) per slice; guarded ratio denominators | Questions 8, 11, 20 |
| Cohorts and retention | Self-joins from a day-0 population; tenure buckets | Questions 2, 16 |
| State transitions and time-in-state | LAG and LEAD per key over ordered events | Questions 10, 16 |
| Broadcast judgment | Name the small side and why it removes the shuffle | Questions 5, 9 |
Warm-ups for the Spark SQL round: one transform, written cleanly, with the boundary condition named before the interviewer asks.
We're auditing the batch pipeline and want the jobs in the overlooked middle by rows processed: not the busiest, not the idlest. Line them up from most rows to least, then return positions 8 through 10, each job's name with its position, lowest position first.
SELECT job_name, rnk
FROM (
SELECT job_name, DENSE_RANK() OVER (ORDER BY rows_done DESC) AS rnk
FROM batch_jobs
) ranked
WHERE rnk BETWEEN 8 AND 10
ORDER BY rnk, job_nameThe core of the Spark SQL round: ranking with ties, cohorts, dedup-latest, pivots, and the broadcast judgment calls.
The developer platform team wants a 7-day token retention curve. A token counts as ACTIVE-WITH-TRAFFIC when its status is 'active' (compare it case-insensitively, since the data mixes 'active' and 'Active') and its requests value is greater than 0. A token with no expiration date is treated as still valid. For each issuance date, report two numbers side by side: how many distinct owners had an active-with-traffic token issued that day (active_day0), and how many of those same owners still held an active-with-traffic token that was valid as of 7 days after that date, i.e. issued on or before issued+7 and not expired before issued+7 (active_day7). Order by the issuance date and return only the first 7 dates.
WITH active_tokens AS (
SELECT token_id, owner_id, issued, expires
FROM api_tokens
WHERE LOWER(status) = 'active' AND requests > 0
)
SELECT a.issued AS the_date,
COUNT(DISTINCT a.owner_id) AS active_day0,
COUNT(DISTINCT b.owner_id) AS active_day7
FROM active_tokens a
LEFT JOIN active_tokens b
ON b.owner_id = a.owner_id
AND DATE(b.issued) <= DATE(a.issued, '+7 days')
AND (b.expires IS NULL OR DATE(b.expires) >= DATE(a.issued, '+7 days'))
GROUP BY a.issued
ORDER BY a.issued
LIMIT 7;
The SRE team is preparing a reliability review for the quarterly infrastructure meeting. Each service has multiple health check records, and the team needs to surface the 10 worst-performing services based on their lowest recorded uptime. If multiple services are tied at the 10th position, include all of them. Return the service name and its lowest uptime value.
SELECT svc_name, min_uptime
FROM (
SELECT
svc_name,
MIN(uptime) AS min_uptime,
DENSE_RANK() OVER (ORDER BY MIN(uptime) ASC) AS rnk
FROM svc_health
GROUP BY svc_name
) ranked
WHERE rnk <= 10
ORDER BY min_uptime ASC
Which device type serves the most unique users based on session data? If there's a tie, include all tied types. Return the device type and user count.
WITH type_users AS (
SELECT
d.device_type,
COUNT(DISTINCT us.user_id) AS user_count
FROM user_sessions us
JOIN devices d ON us.device_id = d.device_id
GROUP BY d.device_type
)
SELECT device_type, user_count
FROM type_users
WHERE user_count = (SELECT MAX(user_count) FROM type_users)
The ad monetization team is evaluating which user demographics drive the most revenue. Show the total ad revenue for each age bucket, with the highest-earning buckets first. Exclude users who have no age bucket on file.
SELECT u.age_bucket,
SUM(ai.revenue) AS total_revenue
FROM ad_impressions ai
INNER JOIN users u
ON ai.user_id = u.user_id
WHERE u.age_bucket IS NOT NULL
GROUP BY u.age_bucket
ORDER BY total_revenue DESC
Return the user IDs and total spend for customers who fall in the top 1% by total transaction amount over the last 7 days.
SELECT user_id, total_spend
FROM (
SELECT
user_id,
SUM(total_amount) AS total_spend,
NTILE(100) OVER (ORDER BY SUM(total_amount) DESC) AS pctl
FROM transactions
WHERE transaction_date >= DATE('now', '-7 days')
GROUP BY user_id
) ranked
WHERE pctl = 1
The finance team wants to track each customer's spending trajectory over time. Show every transaction alongside the customer's cumulative total spend up to and including that row, ordered by transaction date.
SELECT
user_id,
product_id,
total_amount,
transaction_date,
SUM(total_amount) OVER (
PARTITION BY user_id
ORDER BY transaction_date, transaction_id
) AS cumulative_sales
FROM transactions
ORDER BY user_id, transaction_date, transaction_id
The developer experience team is measuring token churn for the platform health dashboard. Compute the fraction of all issued API tokens whose expiration date has already passed, treating tokens with no expiration as still active. Express the result as a decimal ratio, not a percentage.
SELECT CAST(SUM(CASE WHEN expires IS NOT NULL AND date(expires) < date('now') THEN 1 ELSE 0 END) AS REAL) / COUNT(*) AS churn_rate
FROM api_tokens
The data platform team is benchmarking throughput for the highest-priority batch jobs. Among priority-1 jobs, which one processed the most rows? If multiple jobs tie for the top value, include all of them.
SELECT job_id, job_name, rows_done FROM batch_jobs WHERE priority = 1 AND rows_done = (SELECT MAX(rows_done) FROM batch_jobs WHERE priority = 1) ORDER BY job_id
We're auditing our release cadence, where each successful production deploy of a service stays live until the next successful production deploy of that same service replaces it. The environment and status labels were recorded with inconsistent casing over the years, so match them case-insensitively. For every release that was eventually replaced, report the service, the version, and how many calendar days it stayed live, longest-lived first; when two releases held for the same number of days, list them alphabetically by service and then the earlier deploy first.
SELECT
svc_name,
version,
CAST(julianday(date(next_deploy_at)) - julianday(date(deploy_at)) AS INTEGER) AS days_live
FROM (
SELECT
svc_name,
version,
deploy_at,
LEAD(deploy_at) OVER (PARTITION BY svc_name ORDER BY deploy_at) AS next_deploy_at
FROM (
SELECT svc_name, version, deploy_at
FROM deploy_logs
WHERE LOWER(env_name) = 'production' AND LOWER(status) = 'success'
) prod_deploys
) lifespans
WHERE next_deploy_at IS NOT NULL
ORDER BY days_live DESC, svc_name, deploy_at
Our growth team wants to know how far iOS 16, 17, and 18 have spread across our age groups. For each age group, show the number of users who ran a session on one of those iOS versions next to the total number of users with any session, largest groups first.
SELECT u.age_bucket,
COUNT(DISTINCT CASE
WHEN d.os_name = 'iOS'
AND (d.os_version LIKE '16%' OR d.os_version LIKE '17%' OR d.os_version LIKE '18%')
THEN u.user_id
END) AS ios_users,
COUNT(DISTINCT u.user_id) AS total_users
FROM users u
JOIN user_sessions s ON u.user_id = s.user_id
LEFT JOIN devices d ON s.device_id = d.device_id
WHERE u.age_bucket IS NOT NULL
GROUP BY u.age_bucket
ORDER BY total_users DESC, u.age_bucket DESC
The content team is pruning dead pages, where each page_url is already stored in its final canonical form and should be treated exactly as recorded. Find the content with the fewest unique viewers, counting a visitor who returns to the same page many times only once. If several pages share that lowest viewer count, include all of them.
SELECT page_url AS content_id,
COUNT(DISTINCT user_id) AS unique_viewers
FROM page_views
GROUP BY page_url
HAVING COUNT(DISTINCT user_id) = (
SELECT MIN(viewer_count)
FROM (
SELECT COUNT(DISTINCT user_id) AS viewer_count
FROM page_views
GROUP BY page_url
)
)
ORDER BY unique_viewers ASC
The catalog dashboard needs every product that has sold paired with its newest sale, since some products have hundreds of transactions but only the most recent one belongs on the page. Show each product's name and category alongside that latest sale's amount and date.
WITH ranked AS (
SELECT product_id,
total_amount,
transaction_date,
ROW_NUMBER() OVER (
PARTITION BY product_id
ORDER BY transaction_date DESC, transaction_id DESC
) AS rn
FROM transactions
)
SELECT p.product_name,
p.category,
r.total_amount AS latest_sale_amount,
r.transaction_date AS last_sale_date
FROM products p
JOIN ranked r ON r.product_id = p.product_id
WHERE r.rn = 1
ORDER BY p.product_name, p.product_id
Marketing credits each user's acquisition to their first-touch channel: whatever they did first. For each user, return the user_id and that first event's event_type, labeled first_channel.
WITH ranked AS (
SELECT user_id,
event_type,
ROW_NUMBER() OVER (
PARTITION BY user_id
ORDER BY event_timestamp ASC, event_type ASC
) AS rn
FROM event_data
)
SELECT user_id,
event_type AS first_channel
FROM ranked
WHERE rn = 1
ORDER BY user_id;
An operations dashboard spotlights the busiest API endpoints each day. For every day, surface the endpoints whose daily call count is among the three highest counts that day, and report the day, the endpoint, and its level (1 for the busiest count, 3 for the third highest), earliest day first.
WITH ranked AS (
SELECT DATE(call_time) AS call_day,
endpoint,
COUNT(*) AS call_count,
DENSE_RANK() OVER (PARTITION BY DATE(call_time) ORDER BY COUNT(*) DESC) AS rnk
FROM api_calls
GROUP BY DATE(call_time), endpoint
)
SELECT call_day, endpoint, rnk
FROM ranked
WHERE rnk <= 3
ORDER BY call_day ASC, rnk ASC, endpoint ASCThe senior tier: explicit frames, state transitions, time-window joins, and ratio-ordering judgment, with follow-ups on the cost of each.
Compute a 3-month rolling average of total revenue from transactions, excluding refunds (negative amounts). For each month, the average uses the current month and the two preceding months. Show year-month in YYYY-MM format and the rolling average, sorted chronologically. The first two months will not be true 3-month averages.
WITH monthly_rev AS (
SELECT strftime('%Y-%m', transaction_date) AS ym, SUM(total_amount) AS revenue
FROM transactions
WHERE total_amount >= 0
GROUP BY strftime('%Y-%m', transaction_date)
)
SELECT ym, AVG(revenue) OVER (ORDER BY ym ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS rolling_avg
FROM monthly_rev
ORDER BY ym
Each row in deploy_logs is one deployment by an engineer (author), with a status and a deploy_at timestamp. Treat author names case-insensitively (e.g. 'Alice' and 'alice' are the same engineer), and likewise normalize status casing. We want to find engineers who bounced back after a rollback: order each engineer's deployments by deploy_at, and find any deployment whose status is 'rolled_back' where that engineer's very next deployment (by timestamp) has status 'success'. Return a single number: the count of DISTINCT engineers who had at least one such rolled_back -> success transition.
WITH ordered_deploys AS (
SELECT LOWER(author) AS author, LOWER(status) AS status, deploy_at,
LEAD(LOWER(status)) OVER (PARTITION BY LOWER(author) ORDER BY deploy_at) AS next_status
FROM deploy_logs
)
SELECT COUNT(DISTINCT author) AS recovery_count
FROM ordered_deploys
WHERE status = 'rolled_back' AND next_status = 'success'
For each billing entry, compute the cost-per-service ratio (amount divided by the number of services in that region). Then find the monthly average of these ratios. For each year-month, show the average ratio, the monthly average, and the average absolute difference between individual ratios and that month's average.
WITH svc_per_region AS (
SELECT region, COUNT(DISTINCT svc_name) AS svc_count
FROM cloud_costs
GROUP BY region
),
with_ratio AS (
SELECT cc.cost_id, cc.svc_name, cc.region, cc.amount, cc.bill_date,
strftime('%Y-%m', cc.bill_date) AS ym,
CAST(cc.amount AS REAL) / sr.svc_count AS cost_ratio
FROM cloud_costs cc
INNER JOIN svc_per_region sr ON cc.region = sr.region
),
monthly_avg AS (
SELECT ym, AVG(cost_ratio) AS avg_ratio
FROM with_ratio
GROUP BY ym
),
diffs AS (
SELECT wr.ym,
wr.cost_ratio AS actual_ratio,
ma.avg_ratio AS monthly_average,
ABS(wr.cost_ratio - ma.avg_ratio) AS abs_diff
FROM with_ratio wr
INNER JOIN monthly_avg ma ON wr.ym = ma.ym
)
SELECT ym,
AVG(actual_ratio) AS actual_ratio,
AVG(monthly_average) AS monthly_average,
AVG(abs_diff) AS avg_abs_difference
FROM diffs
GROUP BY ym
ORDER BY ym
We're comparing search quality across signup cohorts for a shopping marketplace, grouping users by the calendar year they joined. For each cohort, return how many searches its users ran, how many of those ended in a clicked result, and the resulting success rate.
WITH max_date AS (
SELECT MAX(signup_date) AS latest
FROM users
)
, user_segment AS (
SELECT u.user_id, CASE WHEN julianday((SELECT latest FROM max_date)) - julianday(u.signup_date) <= 30 THEN 'new' ELSE 'existing' END AS segment
FROM users u
)
, search_stats AS (
SELECT us.segment, COUNT(*) AS total_searches, SUM(CASE WHEN sq.clicked_result = 1 THEN 1 ELSE 0 END) AS successful_searches
FROM search_queries sq
INNER
JOIN user_segment us ON sq.user_id = us.user_id
GROUP BY us.segment
)
SELECT segment, total_searches, successful_searches, CAST(successful_searches AS DOUBLE) / total_searches AS success_rate
FROM search_stats
Compare total transaction amounts in Q4 vs Q3 to compute quarter-over-quarter revenue growth by region. Growth is ((Q4 total minus Q3 total) / Q3 total) * 100. Only include regions with sales in both quarters. Return the region and growth percentage.
WITH q3 AS (
SELECT user_id AS region, SUM(total_amount) AS total
FROM transactions
WHERE CAST(strftime('%m', transaction_date) AS INTEGER) BETWEEN 7 AND 9
GROUP BY user_id
)
, q4 AS (
SELECT user_id AS region, SUM(total_amount) AS total
FROM transactions
WHERE CAST(strftime('%m', transaction_date) AS INTEGER) BETWEEN 10 AND 12
GROUP BY user_id
)
SELECT q4.region, (q4.total - q3.total) * 100.0 / q3.total AS growth_pct
FROM q4
JOIN q3 ON q4.region = q3.region12 production incidents to diagnose. A pager alert, the Spark UI evidence tabs (Stages, Executors, Logs, Plan), and a live editor for the fix. This is the diagnosis half of the Spark round.
Spark rounds converged on incidents because API questions stopped separating candidates. Anyone prepping for a week can write a groupBy; what employers need to know is what you do when the job that ran in 22 minutes every night for a year is suddenly 4 hours into a stage with 1 task still running. That situation is the actual job. Skew from a hot key, an OOM from a wide operator, a scan that quietly stopped pruning, a stream double-counting after a broker restart: these are the handful of failure classes every production Spark deployment meets eventually, which is why the same scenarios surface in loop after loop, from Databricks to Uber, nearly verbatim.
The questions look the way they do because the Spark UI is the entire interface of a real diagnosis. A pager line, a task table, executor stats, logs, a plan: that is what on-call hands you, so that is what the interviewer hands you. The skill being measured is reading the numbers out loud: 199 tasks finishing in seconds beside 1 running for 3 days is skew; shuffle write dwarfing input is a wide operator materializing too early; partitions read equal to the whole table is a filter the planner cannot see. Each scenario below ships that evidence exactly as the UI would show it, a diagnosis to commit to, and the job's code to fix, because the rubric is the same one on-call uses: say why it broke, prove it from the evidence, then fix it without breaking the contract downstream.
Your team's text analytics pipeline runs a word count job over a 50 GB corpus every night. It has been working fine for months, but after the corpus grew 3x last quarter the job started failing. The Spark UI shows 48 GB of shuffle write and three executors dead from OOM. The code uses groupByKey. Fix it.
from pyspark import SparkContext
sc = SparkContext.getOrCreate()
# groupByKey materializes all values per key before aggregating -> OOM on high-freq words
# reduceByKey applies a partial combiner on the map side before shuffling -> 10x less data moved
text = sc.textFile("s3://apple-ml-data/nlp-corpus/")
word_counts = (
text
.flatMap(lambda line: line.lower().split())
.map(lambda word: (word, 1))
.reduceByKey(lambda a, b: a + b) # partial sum per partition before shuffle
)
word_counts.saveAsTextFile("s3://apple-ml-data/word-counts-output/")Why this matters. groupByKey ships every value across the shuffle and materializes 140M objects for a stop-word before any aggregation runs; reduceByKey folds map-side first, so the wire carries partial sums instead of raw occurrences. The interviewer cares whether you reason from the 48 GB shuffle-write number to the operator, not whether you memorized the rule.
A client's daily export pipeline reads 200 GB of transaction data, filters it to about 2 GB of flagged records, and writes Parquet to S3. Downstream Athena queries on this table are taking 45 seconds for a simple COUNT(*). You check S3 and find 2,000 Parquet files averaging 1 MB each. The job has spark.sql.shuffle.partitions set to 2000. Fix the write so Athena can actually query this table.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.getOrCreate()
# After filter: 2 GB of data. Target file size ~200 MB -> 10 files.
# coalesce(N) merges partitions without a full shuffle (narrow dependency).
# repartition(N) would trigger a full shuffle - unnecessary here since we
# are only reducing partition count, not redistributing keys.
flagged = (
spark.table("clickstream_raw")
.filter(col("is_flagged") == True)
.coalesce(10) # 2 GB / 10 = 200 MB per file, Athena-friendly
)
flagged.write .mode("overwrite") .parquet("s3://exports/flagged_transactions/")Why this matters. 2,000 output files means 2,000 final partitions. coalesce merges them without a shuffle on the way out; repartition would pay a full shuffle for evenness nobody asked for. Reading the file count as a partition count, then picking the no-shuffle knob, is the whole diagnosis.
The order enrichment job joins a 500M-row retail_orders table (80 GB) against a 5,000-row stores dimension (30 MB) on store_id. The join takes 12 minutes and shuffles 80 GB. The physical plan shows SortMergeJoin with Exchange (shuffle) on both sides. The stores table is 30 MB. Why did Spark choose SortMergeJoin, and how do you fix it?
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast
spark = SparkSession.builder.getOrCreate()
retail_orders = spark.table("retail_orders") # 500M rows, 80 GB
stores = spark.table("stores") # 5,000 rows, 30 MB
# Option 1: broadcast() hint - overrides autoBroadcastJoinThreshold for this join
result = retail_orders.join(broadcast(stores), "store_id", "inner")
# Option 2 (cluster-wide): raise threshold to cover this table
# spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50m")
result.select("order_id", "amount", "store_name", "region") .write.mode("overwrite").saveAsTable("order_enriched")Why this matters. df.explain() answers this before any config does: the plan names the scan, the exchange, and the join strategy, and one of them is the bottleneck. The skill is walking the plan bottom-up and pricing each exchange, because every tuning question afterward is this question with the plan hidden.
A daily analytics job reads a 3 TB user_events Parquet table partitioned by event_date, filters to yesterday (about 10 GB), and joins against user_profiles. The job takes 40 minutes but should take 5. A colleague wrote the pipeline using a subquery pattern that defeats partition pruning. The physical plan shows a full table scan of all 3 TB. Rewrite the query so Catalyst pushes the date filter down to the file scan.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, date_sub, current_date
spark = SparkSession.builder.getOrCreate()
yesterday = date_sub(current_date(), 1)
# Column expression allows Catalyst to push predicate to PartitionFilters at FileScan.
# A scalar subquery or Python variable passed through withColumn does NOT push down.
df = (
spark.table("user_events")
.filter(col("event_date") == yesterday)
)
result = (
df.groupBy("user_id", "country")
.count()
)
result.write.mode("overwrite").saveAsTable("user_events_daily")Why this matters. Catalyst pushes filters it can see: a Column-expression predicate reaches the scan, a filter behind a UDF or an opaque cast does not. The evidence shows the same query with 2 wildly different scan sizes; the fix is rewriting the predicate so pushdown applies, not adding memory to survive reading everything.
An iterative ML feature engineering pipeline reads a 200 GB base DataFrame and runs 8 sequential enrichment steps. Each step joins against a different dimension table and adds columns. A previous engineer cached the base DataFrame to speed up the repeated reads, but after step 4 executors start dying with OOM. The cache is eating so much memory that later steps have no room for shuffle data. Fix the caching strategy so the pipeline completes without OOM.
from pyspark.sql import SparkSession
from pyspark.storagelevel import StorageLevel
spark = SparkSession.builder.getOrCreate()
base_df = spark.table("ml_base_features").persist(StorageLevel.MEMORY_AND_DISK)
dims = [spark.table(f"dim_{i}") for i in range(1, 9)]
current_df = base_df
for i, dim in enumerate(dims):
prev_df = current_df
current_df = current_df.join(dim, "user_id", "left")
# Unpersist the previous step immediately - only base_df stays pinned
if prev_df is not base_df:
prev_df.unpersist()
current_df.write.mode("overwrite").saveAsTable("ml_enriched_features")
base_df.unpersist()Why this matters. An iterative loop that caches a new DataFrame each epoch without unpersisting the last one grows memory forever; the evidence's climbing storage tab is the tell. cache() is a contract with the executor's memory, and the fix pairs every cache with an unpersist once the next iteration's lineage no longer needs it.
A Spark 3.4 job joins a 400 GB search_logs table against a 60 GB ad_clicks table on query_id. Takes 90 minutes. Spark UI shows moderate skew: the top partition has 8x the median row count. A colleague suggests salting, but the codebase is complex and salting would require changes in three downstream jobs. Enable and configure Adaptive Query Execution to let Spark handle the skew at runtime, coalesce small partitions, and optimize the join strategy automatically.
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
# Enable AQE: runtime skew handling, small-partition coalescing, dynamic join strategy
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
# Treat partitions > 5x median as skewed; split at 256 MB advisory target
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "256m")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
search_logs = spark.table("search_logs")
ad_clicks = spark.table("ad_clicks")
result = search_logs.join(ad_clicks, "query_id", "inner")
result.write.mode("overwrite").saveAsTable("search_ad_attribution")Why this matters. AQE splits skewed partitions it can measure at shuffle time, and the evidence shows it doing so, up to the boundary it cannot see past: a skew created inside one partition by the join key itself. Knowing what AQE fixed automatically and what still needs a salt is the modern version of every skew question.
A Spark job building daily product recommendation features keeps failing with different errors depending on the cluster config. With 2 large executors (64 GB each, 16 cores), the job dies from GC pauses. When a colleague tried 32 small executors (4 GB each, 1 core), broadcast joins fail because the 2 GB broadcast variable does not fit. Find a balanced executor configuration for a 50-node cluster with 128 GB RAM and 32 cores per node.
# Sizing rationale for 50-node cluster (128 GB RAM, 32 cores/node):
#
# Cores per executor: 4-5 sweet spot. Avoids GC thrash (>5 threads/JVM),
# keeps HDFS throughput (3-5 concurrent tasks per executor optimal).
#
# Executors per node: floor((128 - 1 OS) / (18 + 2 overhead)) = 6
# 1 executor reserved for YARN NodeManager per node.
#
# Total executors: 50 * 6 - 1 (YARN AM) = 299 ~ 300
# Total cores: 300 * 4 = 1200
#
# Memory: 18 GB heap + 2 GB overhead = 20 GB per executor
# Broadcast variable (2.1 GB) fits comfortably in 18 GB heap.
# spark.memory.fraction=0.6 -> 10.8 GB for execution/storage, 7.2 GB for user data.
spark.executor.memory=18g
spark.executor.cores=4
spark.executor.instances=300
spark.executor.memoryOverhead=2g
spark.sql.autoBroadcastJoinThreshold=3g # raise threshold to allow 2.1 GB broadcastWhy this matters. The 2 failures bracket the answer: 16 cores per JVM thrashes GC because every thread fights one heap, and a 4 GB executor cannot host a 2.1 GB broadcast once JVM overhead is added. Derive the 4-to-5-core, mid-size executor from those constraints, state memoryOverhead explicitly, and the question is done.
A nightly job refreshes daily_category_sales, a summary of one day's sales broken down by product category, but it rebuilds from the entire multi-year transactions history on every run even though only the most recent day is ever new. For the latest day present in transactions, total each product category's revenue and units sold, and make the job read only that day's data rather than the whole table.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, current_date, date_sub
spark = SparkSession.builder.getOrCreate()
yesterday = date_sub(current_date(), 1)
# Partition filter pushed down to FileScan - reads 1 partition, not 1,840
df = (
spark.table("store_sales_raw")
.filter(col("sale_date") == yesterday)
)
pivoted = (
df.groupBy("store_id", "sale_date")
.pivot("product_category", ["grocery", "electronics", "apparel", "home", "beauty"])
.agg(_sum("revenue").alias("revenue"), _sum("qty").alias("qty"))
)
pivoted.write.mode("overwrite").saveAsTable("daily_store_sales")You are the on-call data engineer at a streaming company. The nightly viewing_engagement Spark job just paged you. It normally finishes in 45 minutes but has been running for over two hours and is still stuck. The job joins a large event_data table (800M rows/day of viewing, playback, and interaction events) against a small app_users dimension (2M subscribers) to produce daily engagement metrics by event type and account tier. Your SLA is 60 minutes. Diagnose the root cause using the Spark UI evidence and fix the job so it meets SLA.
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col
spark = SparkSession.builder.getOrCreate()
event_data = spark.table("event_data") # 800M rows, skewed on user_id
app_users = spark.table("app_users") # 2M rows, 48 MB
# Root cause: SortMergeJoin chosen because app_users (48 MB) > autoBroadcastJoinThreshold (10 MB).
# Fix: broadcast() hint bypasses the threshold. app_users sent to each executor once;
# event_data never shuffled -> free_tier skew is irrelevant (no hash partitioning step).
result = (
event_data
.join(broadcast(app_users), "user_id", "inner")
.groupBy("event_type", "account_tier")
.count()
)
result.write.mode("overwrite").saveAsTable("viewing_engagement")Why this matters. Power users concentrate the join: the task table shows a handful of tasks carrying the stage. The escalation ladder is the answer: broadcast the small side if it fits, and when both sides are large, split the hot keys out, salt them, and union the rest. Jumping straight to salting everything doubles the work for the cold keys.
The daily payment reconciliation Spark job joins 1.2 billion payments against a 500K-row merchants dimension on merchant_id. It has been failing for three days. Spark UI shows one task processing 38% of all rows while the other 199 finish in seconds. The hot merchant is your company's internal payment processor that handles all driver payouts. You cannot broadcast merchants because a downstream join adds a 2 GB enrichment table. Propose and implement a salting strategy.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, rand, floor
spark = SparkSession.builder.getOrCreate()
SALT_FACTOR = 50 # splits hot key across 50 partitions
payments = spark.table("payments")
merchants = spark.table("merchants")
# Add random salt [0, SALT_FACTOR) to each transaction row
txn_salted = payments.withColumn(
"salt", floor(rand() * SALT_FACTOR).cast("int")
)
# Cross-join merchants against the SALT_FACTOR salt values, one copy each
salts = spark.createDataFrame([(i,) for i in range(SALT_FACTOR)], ["salt"])
merchants_exploded = merchants.crossJoin(salts)
# Join on composite key - UBER_INTERNAL splits across 50 partitions
result = (
txn_salted
.join(merchants_exploded, ["merchant_id", "salt"], "inner")
.drop("salt")
)
result.write.mode("overwrite").saveAsTable("payment_reconciled")Why this matters. One merchant is 38% of 1.2B rows and broadcasting is off the table, so salt: explode the dimension across a salt range, scatter the fact side with floor(rand() * SALT), join on the composite key, re-aggregate. State the cost unprompted: the dimension multiplies by the salt factor, which is why production code salts the detected hot keys, not the whole join.
A Structured Streaming job reads click events from Kafka, joins against a user dimension, and writes aggregated metrics to Delta Lake every 2 minutes. After a Kafka broker restart last week, the consumer group replayed 15 minutes of events, creating duplicate click counts in the output. The business team noticed inflated metrics for that window. Add watermark-based deduplication so that late or replayed events within a 30-minute window are dropped.
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col, window
spark = SparkSession.builder.getOrCreate()
clicks = (
spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "kafka:9092")
.option("subscribe", "click_events")
.option("startingOffsets", "latest")
.load()
.selectExpr(
"CAST(value AS STRING) AS json",
"CAST(key AS STRING) AS click_id",
"timestamp AS event_time",
)
.selectExpr("click_id", "event_time",
"get_json_object(json, '$.user_id') AS user_id",
"get_json_object(json, '$.page_id') AS page_id")
)
app_users = spark.table("app_users") # 2M rows - broadcastable
deduped = (
clicks
.withWatermark("event_time", "30 minutes")
# dropDuplicatesWithinWatermark (Spark 3.5+): state is bounded by watermark.
# Falls back to dropDuplicates(["click_id"]) on Spark < 3.5 (unbounded state - monitor size).
.dropDuplicatesWithinWatermark(["click_id"])
.join(broadcast(app_users), "user_id", "inner")
.groupBy(
window(col("event_time"), "2 minutes"),
"account_tier"
)
.count()
.withColumnRenamed("count", "click_count")
)
query = (
deduped.writeStream
.format("delta")
.outputMode("append")
.option("checkpointLocation", "s3://checkpoints/click_metrics/")
.trigger(processingTime="2 minutes")
.toTable("click_metrics")
)
query.awaitTermination()Why this matters. At-least-once replay is Kafka's contract, not an incident. Exactly-once output means an event-time watermark plus dropDuplicates on the event key inside that watermark, writing to an idempotent Delta sink. The follow-up is always what bounds the state store, and the watermark is the answer; without it the dedup state grows forever.
A PySpark pipeline scores 2 billion credit payments per day for fraud using a Python UDF. The UDF computes a risk score from 12 columns: thresholds, lookups, weighted sums. All expressible with native Spark functions. The job takes 4 hours. Profiling shows 70% of the time in ArrowEvalPython (serializing data between JVM and Python). Rewrite the UDF as native Spark SQL expressions.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, lit
spark = SparkSession.builder.getOrCreate()
# Lookup tables as broadcast constants (no UDF needed)
COUNTRY_RISK = {"US": 0.1, "NG": 0.9, "BR": 0.6, "IN": 0.4, "GB": 0.2}
MERCHANT_TIER_WEIGHT = {1: 0.05, 2: 0.15, 3: 0.3, 4: 0.6}
txns = spark.table("payments")
# Replace all 12 UDF columns with native Spark expressions.
# Catalyst optimizes the expression tree: constant folding, whole-stage codegen.
# Zero JVM-to-Python serialization.
risk_score = (
when(col("is_international"), 0.35).otherwise(0.0)
+ when(col("channel_risk") == "card_not_present", 0.25).otherwise(0.0)
+ when(col("velocity_1h") > 10, 0.20).otherwise(col("velocity_1h") * 0.02)
+ when(col("velocity_24h") > 50, 0.15).otherwise(col("velocity_24h") * 0.003)
+ when(col("device_trust") < 0.3, 0.15).otherwise(0.0)
+ when(col("ip_risk") > 0.7, 0.10).otherwise(col("ip_risk") * 0.14)
+ when(col("card_age_days") < 30, 0.10).otherwise(0.0)
+ when(col("txn_hour").isin([0, 1, 2, 3]), 0.08).otherwise(0.0)
+ when(col("days_since_last_txn") > 180, 0.05).otherwise(0.0)
+ when(col("amount") > 5000, 0.10).otherwise(col("amount") / 50000)
)
result = (
txns
.withColumn("risk_score", risk_score.cast("double"))
.withColumn("flagged", col("risk_score") > 0.8)
.select("txn_id", "merchant_id", "risk_score", "flagged")
)
result.write.mode("overwrite").saveAsTable("fraud_scores")Why this matters. A Python UDF round-trips every row across the JVM boundary and blinds the optimizer on top. The logs show the serialization tax directly. Rewriting with native column functions keeps the work in the JVM and restores pushdown; a pandas UDF is the halfway house when native ops cannot express the logic.
The verbal questions between coding prompts. Spark rounds lean on the execution model, not API trivia.
A shuffle repartitions data across the cluster so rows with the same key land together: every wide operation (groupBy, join, distinct, repartition) pays for one. It is the unit of cost in Spark: disk, network, and serialization all spike at shuffle boundaries, which is why plans are read shuffle-first.
reduceByKey combines map-side before shuffling, so the wire carries partial aggregates; groupByKey ships every raw value and materializes them per key on one executor. Same output for associative aggregations, wildly different memory profile. In the DataFrame API, agg() gets the combiner behavior automatically.
repartition shuffles to exactly N partitions, up or down, evenly. coalesce merges existing partitions without a shuffle, down only, and can leave them uneven. Writing fewer output files wants coalesce; rebalancing skew before an expensive stage wants repartition, and paying the shuffle there is the point.
When one side fits comfortably in each executor's memory, typically under a few hundred MB after compression. The small side ships to every executor once and the big side never shuffles. The failure mode is broadcasting something that grew: yesterday's 100 MB dimension is next quarter's 4 GB OOM.
One key holding a disproportionate share of rows, so one partition and one task do most of the work. The tell is a stage where the max task duration dwarfs the median while shuffle-read is similarly lopsided. Fixes in escalating order: filter the degenerate key, broadcast the other side, salt the hot keys.
Adaptive Query Execution replans at runtime using real shuffle statistics: it coalesces small partitions, flips sort-merge to broadcast when a side turns out small, and splits skewed partitions. It does not remove the need to understand the plan; it removes the need to hand-tune the cases it can see.
Each row round-trips between the JVM and a Python worker with serialization both ways, and the optimizer treats the UDF as a black box, killing pushdown. Native column functions stay in the JVM and stay optimizable; pandas UDFs amortize the boundary with Arrow batches when native ops cannot express the logic.
3 parts together: replayable sources (Kafka offsets in the checkpoint), deterministic state bounded by a watermark, and idempotent or transactional sinks (Delta). Any one alone is not enough, which is what makes duplicate counts such a common production failure.
Skew is one join or group key holding a disproportionate share of rows, so a single task processes most of the data while the rest of the cluster idles and the stage hangs at 99%. Spot it in the Spark UI as a task whose shuffle-read bytes dwarf the median. Fixes in order of preference: broadcast the small side to remove the shuffle, enable AQE skew join handling, or salt the hot key with a random suffix, join on the salted key, then re-aggregate.
Both cut data before it is read. Predicate pushdown sends the filter down into the file scan, so Parquet row groups whose min/max statistics cannot match are skipped. Partition pruning goes further and skips whole directories when the filter is on the partition column. Both are defeated by wrapping the column in a function or by a filter the optimizer cannot see through, such as one hidden inside a UDF.
Catalyst rewrites the query through phases: parse to an unresolved logical plan, resolve against the catalog, optimize with rule-based transformations (predicate pushdown, constant folding, column pruning), then generate physical plans and pick one by cost. The practical consequence is that Python UDFs are opaque boxes it cannot rewrite or push down, which is why replacing one with a built-in expression often changes the whole plan.
A narrow transformation (map, filter, union) needs only the partition it is given, so it pipelines within a stage with no data movement. A wide transformation (groupBy, join, distinct, repartition) needs rows from other partitions, which forces a shuffle and a stage boundary. Counting stage boundaries in a plan tells you how many shuffles a job pays for.
cache() is persist() with the default MEMORY_AND_DISK storage level; persist() lets you name a different level. Use them when a DataFrame is consumed more than once and recomputing it means re-reading the source or redoing a shuffle. Caching something read once wastes memory and can evict work that mattered. Unpersist when the reuse window closes.
Executor memory splits into an execution region (shuffles, joins, sorts) and a storage region (cached blocks), which borrow from each other, plus user memory and a fixed overhead. OOM usually means one of: a skewed partition too large for one task, a collect() pulling the result set to the driver, an oversized broadcast, or too little overhead for Python worker processes in PySpark. The fix follows the cause; raising memory blindly hides skew rather than solving it.
Start at the Stages tab and sort by duration. Open the slowest stage and compare the task duration and shuffle-read distribution: a max far above the median is skew, uniformly slow tasks with heavy spill is memory pressure, and a huge task count on a small input is over-partitioning. The SQL tab then maps that stage back to the plan node so you know which operator to change.
RDD is the low-level distributed collection with no schema, so Catalyst cannot optimize it. DataFrame is a Dataset of Row with a schema, fully optimized, and the right default. Dataset adds compile-time typing but exists only in Scala and Java. In PySpark the choice is effectively DataFrame, and dropping to RDD is a deliberate decision that gives up the optimizer.
Caching keeps a computed result available for reuse but preserves the lineage, so a lost partition is recomputed. Checkpointing writes the data to reliable storage and truncates the lineage entirely. Long iterative jobs checkpoint to stop the lineage graph from growing without bound; streaming jobs checkpoint to store offsets and state so a restart resumes exactly where it stopped.
Thousands of tiny output files make every downstream read pay per-file listing and open overhead, and they bloat the metastore. It usually comes from over-partitioning or from a streaming sink writing per micro-batch. Fixes: coalesce before writing, partition on a lower-cardinality column, or run a compaction job. On Delta or Iceberg, OPTIMIZE or a rewrite action does the compaction for you.
From reported debriefs, these are the recurring failure modes, not API syntax.
Raising executor memory on a skew problem burns the round. The task table says which failure class you are in; read it out loud first. Interviewers reward the diagnosis order more than the fix.
broadcast() on a side you have not sized is the classic self-inflicted OOM. Say the size, say the executor memory, then broadcast. If you cannot say the size, that is the follow-up.
A filter buried in Python logic or applied after a UDF reads the whole table. Partition filters must be Column expressions at the scan. Check partitions-read in the plan, not the WHERE clause in the code.
Salting multiplies the dimension by the salt factor. Applied to the whole join instead of the detected hot keys, it turns one problem into two. Name the hot keys, salt those, union the rest.
Most Spark rounds open from evidence, not from a blank editor: a Spark UI screenshot, a task table, a plan, or a written incident report. The first minutes are diagnosis, and the rubric rewards naming the failure class from the numbers before proposing anything. Candidates who jump to config changes without reading the evidence fail the round in the first 5 minutes.
The coding half is smaller than candidates expect: a dozen lines of DataFrame API once the diagnosis is right. What gets probed is the why behind each line: why broadcast here, why repartition before the write, why the watermark bounds the state. The SQL round's aggregation and window patterns all reappear in DataFrame syntax, so that preparation compounds.
Seniority moves the bar from mechanics to contracts: exactly-once semantics, schema evolution on the sink, what replaying yesterday does to downstream consumers. If the answer to a failure is idempotent by construction, most follow-ups answer themselves.
Reading a solution is not the same as writing one
Every engineer who has frozen on a query they had read a dozen times knows the gap. The only preparation that closes it is producing the answer yourself, under time, before the interview does it for you
76% of hiring managers reject on the coding task, not the resume
From HackerRank's 2024 Developer Skills Report. Candidates who look strong on paper still fail the live screen if they haven't done timed, executable practice
Spark rounds are diagnosis rounds
Skew, shuffles, broadcast judgment, partition pruning, exactly-once streaming. Reading the task table and naming the failure class is the signal, not the API
The SQL round's patterns on Spark: windows, aggregation, MERGE, and broadcast hints.
The DataFrame API surface: transforms, joins, and the Catalyst boundary.
Plans, partitions, caching, and the failure classes senior rounds probe.
The full Spark practice catalog: DataFrame transforms, joins, and tuning.
The Spark-first companies' question shapes, end to end.
Delta, AQE, and the platform-side questions Databricks loops add.
The same patterns in their home dialect: 20 solved, one per company, none repeated here.
Every round in one place: SQL, Python, Spark, data modeling, and pipeline design.
The same questions, asked the way an interviewer asks them: vague, timed, with follow-ups.