943 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 943 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 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. |
| 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. |
| Broadcast joins | 13% (2) | 3 | Replacing a shuffle join with a broadcast when one side fits in memory. |
| 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; surface the 10 least reliable, judged by their lowest recorded uptime, and if several services are tied for that 10th spot include all of them. List each service with its lowest uptime, weakest first.
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
The device team's reach report crowns a single winner, but the sessions in user_sessions show other device types serving nearly as many unique users. Using each session's device type from devices, list every device type that reaches at least 90% of the leader's unique-user count, along with that 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)
How much ad revenue does each age bucket bring in, with the impressions' revenue in ad_impressions and each user's bucket in users, from the highest-earning bucket down? Users with no age bucket on file stay out of the totals.
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
The bar for the top third is not a dollar figure: it sits wherever this window's buyers land against each other, each judged on everything they paid over the last 90 days of transactions rather than on any single purchase. Return the customers above that bar with each one's total, biggest spenders first.
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
Every customer shows up as one lifetime spend figure in the finance team's review, and nobody can see how that figure built up purchase by purchase. From the transactions table, list every purchase beside that customer's cumulative spend up to and including it, customers by ascending user_id and each customer's purchases oldest first, with same-day purchases taken in transaction_id order.
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
Compute the churn rate of the issued tokens in api_tokens as a decimal fraction between 0 and 1, for the platform health dashboard. A token has churned once its expiration date is in the past; a token with no expiration date never lapses, so it stays active while still counting as issued.
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 profiling batch_jobs for throughput leaders: return every job that processed at least 95% as many rows as the single busiest job managed.
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(DATEDIFF('day', date(deploy_at), date(next_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 the latest phone releases, iOS or Android 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 releases next to the total number of users with any session.
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
Find every page in page_views whose unique viewer count is within one of our least-viewed page's, fewest viewers first: that whole tail is our pruning shortlist. Our URL paths are case-sensitive, so /Home and /home are separate pages.
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
An analyst reviewing the catalog dashboard found sale figures that matched no single sale, when only each product's most recent transaction in transactions belongs on the page and products that never sold have nothing to show. List those products alphabetically by name from products, with each one's category and the amount and date of its latest sale.
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
app_event_deliveries records every arrival of an app event, so an event the pipeline redelivered appears again under the same event_uuid and event_ts, and a user's first touch is the event_name of their earliest event. Return each user's user_id exactly once with that name as first_event; deliveries carrying no user_id belong to nobody.
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;
Our daily busiest-endpoints panel no longer matches what on-call sees in api_calls, partly because some endpoint paths were logged with stray spaces around them and show up as two different endpoints. For every day, return the endpoints whose call count is among that day's three highest counts, with the day, the cleaned endpoint, and its level (1 for the busiest count, 3 for the third highest).
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.
Give the finance team a 3-month rolling average of monthly revenue from the transactions table, oldest month first, where each month's figure averages its own revenue total with the two months before it. Refunds are logged there as negative amounts and are not revenue, and the first two months average over only the history they have.
WITH monthly_rev AS (
SELECT TO_CHAR(transaction_date, 'YYYY-MM') AS ym, SUM(total_amount) AS revenue
FROM transactions
WHERE total_amount >= 0
GROUP BY TO_CHAR(transaction_date, 'YYYY-MM')
)
SELECT ym, AVG(revenue) OVER (ORDER BY ym ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS rolling_avg
FROM monthly_rev
ORDER BY ym
In deploy_logs, a rolled-back deployment counts as a bounce-back for the release team when that same engineer's very next deployment succeeds, and both the author names and the status values were written in whatever casing the tooling happened to use, which means nothing. For each service, report how many bounce-backs started on it and how many different engineers pulled one off.
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'
Some months' per-service cost figures swing far more than others, and the finance team's chargeback review wants that spread measured straight from cloud_costs. Take each billing line at its whole-dollar amount with the cents simply dropped, then split it evenly across the number of different services billed in its region to get a decimal cost ratio. For every year-month from the earliest forward, show the average ratio, the month's mean, and the average absolute gap between each line's ratio and that mean.
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,
TO_CHAR(cc.bill_date, 'YYYY-MM') 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
A search only counts as a success when the user clicked one of its results, and on our shopping marketplace everyone in users who signed up in the same calendar year forms one cohort. List the cohorts from earliest to most recent with how many search_queries their users ran, how many of those succeeded, and the success rate.
WITH max_date AS (
SELECT MAX(signup_date) AS latest
FROM users
)
, user_segment AS (
SELECT u.user_id, CASE WHEN DATEDIFF('day', u.signup_date, (SELECT latest FROM max_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
A region only gets a growth figure if it sold in both Q3 and Q4; one with sales in just one of those quarters is left out. We key regions by user_id in transactions and pool every year's July to September sales as Q3 and October to December sales as Q4. Report each region's Q4 total as a percentage change over its Q3 total.
WITH q3 AS (
SELECT user_id AS region, SUM(total_amount) AS total
FROM transactions
WHERE CAST(TO_CHAR(transaction_date, 'MM') 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(TO_CHAR(transaction_date, 'MM') 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.
Ever since the input tripled, the nightly tally of calls per response status that an on-call engineer owns has died with 48 GB of shuffle write and three executors out of memory, because it collects every single call for a status code onto one machine before counting them. She needs the same per-status call totals from api_calls, produced in a way that stays small no matter how many calls pile onto the busiest status code.
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/")A daily export pipeline reads 200 GB of transaction data, keeps only the flagged high-value records (transactions over $500), and writes them as Parquet to S3. Downstream Athena queries on the exported table take 45 seconds for a simple COUNT(*), and S3 holds 2,000 Parquet files averaging 1 MB each even though the job already sets spark.sql.shuffle.partitions to 2000. Read the transactions table, keep the flagged records, and fix the write so the output lands as a handful of right-sized files.
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/")A 30 MB stores dimension should never cost an 80 GB shuffle, yet the platform team's order enrichment plan puts an Exchange on both sides of a SortMergeJoin and runs 12 minutes against a 4-minute SLA. Work out why the planner passed on the cheaper strategy, then rebuild the enrichment from retail_orders and stores so the orders side never moves, keeping each order's id and amount with its store's name and region.
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")user_events is stored as one folder per event_date, and the growth team's 2026 activity rollup has been reading every year of history because its date bound was computed in a separate lookup the file scan never sees. Rebuild the rollup with the bound written straight against event_date: one row per country with the number of events since the start of 2026.
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")For each user in the transactions table, what is their total spend, how many transactions did they make, and what percentage of everyone's combined spend do they account for? Every one of those answers reads that same table, which in production runs to hundreds of millions of rows, so we want it pulled from storage only once and released after its last use, with the profiles coming back in any order.
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()The nightly match of transactions against users on user_id has crept to 90 minutes against a 30-minute SLA, and the on-call engineer traced it to one merge task reading 8x the median rows on a Spark 3.4 cluster whose job config switched adaptive execution off. Salting the key would ripple into three downstream jobs, so she wants a configuration-only fix that splits the hot slice at runtime while the match keeps its current merge strategy and still returns each matched transaction's user_id, transaction_date and total_amount.
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")On a 50-node cluster where every machine has 128 GB and 32 cores, keeps 1 core and 1 GB for the OS, and splits its remaining memory in whole gigabytes among as many executors as its free cores allow, each executor gives up 2 GB of that share to off-heap overhead and keeps the rest as heap. The recommendations team's feature job already died on 16-core executors from GC thrash and on 4 GB executors that could not hold its 2.1 GB broadcast table, so only 1 to 5 cores per executor are in play; with no cluster table to read, number the rows of users and take positions 1 through 5 as those core counts. List the sizings whose heap is at least three times the broadcast size, with cores per executor, executors per node, heap in GB, and the executor count across the whole cluster.
# 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 broadcast
The nightly daily_category_sales job reads every date partition of transactions, yet the summary it rebuilds only covers sales dated on or after 2026-07-01, so the on-call engineer has to make the read touch that slice and nothing earlier. From those sales, produce each product category's total revenue as recorded on the sale and total units sold, with the category taken from products.
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")Two hours into a nightly viewing_engagement run that normally takes 45 minutes against a 60-minute SLA, the on-call engineer is staring at a job that matches the 800M-row-a-day event_data table to the 2M-subscriber users dimension on user_id and counts events per event type and account status. From the Spark UI evidence, find the root cause and fix the job so the large event table is never moved across the cluster, leaving the counts unchanged.
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")transactions holds one payment per row keyed by user_id, and a single internal payout account owns 38% of them, so matching them to users on that key parks the whole account on one task for three days while the other 199 finish in seconds. Without broadcasting users (a downstream 2 GB enrichment step rules that out), spread that account's rows across many tasks and return every account in users that has payments, with its payment count and total amount.
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")Since the broker restart, our consumers keep receiving the same event again under a fresh msg_id, so the only identity that survives a replay is the producer's key in part_key, and a message that arrived without one still counts as a delivery but matches no event. From stream_msgs, show each named consumer's total deliveries next to how many unique events it actually received.
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()Each row of payments is one card transaction, and the starter's Python risk_score UDF scores it from ten of its columns through thresholds, a night-hours check and weighted sums, with a transaction flagged when its raw score clears FLAG_CUTOFF. Return every transaction's txn_id, merchant_id, risk_score rounded to three decimals and flagged, scoring exactly as the UDF does but with no Python left in the plan, since at two billion rows a day the JVM-to-Python round trip is 70% of the job's four hours.
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")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
Spark SQL interview questions
The SQL round's patterns on Spark: windows, aggregation, MERGE, and broadcast hints.
Spark DataFrame questions
The DataFrame API surface: transforms, joins, and the Catalyst boundary.
Spark optimization questions
Plans, partitions, caching, and the failure classes senior rounds probe.
PySpark practice problems
The full Spark practice catalog: DataFrame transforms, joins, and tuning.
Spark data engineer problems
The Spark-first companies' question shapes, end to end.
Databricks interview problems
Delta, AQE, and the platform-side questions Databricks loops add.
SQL interview questions (the same loop)
The same patterns in their home dialect: 20 solved, one per company, none repeated here.
Full data engineer question catalog
Every round in one place: SQL, Python, Spark, data modeling, and pipeline design.
Spark mock interviews
The same questions, asked the way an interviewer asks them: vague, timed, with follow-ups.
The data engineering challenge
Dirty, production-shaped data, scored blind each week.
50+ guides covering every round, company, and role