Databricks Spark Runtime
Answer First: Spark fundamentals still determine Databricks performance: reason from the logical plan to stages, shuffles, tasks, executors, and observable evidence in the Spark UI.
Memory Map: plan -> stages -> shuffle -> tasks -> executors -> Spark UI.
Delta ownership: This module keeps Databricks integration context concise. Transaction-log, MERGE, time-travel, CDF, and maintenance internals are canonical on /learn/delta.
Spark Architecture & Internals
💡 Interview Tip
Focus: Deep internals, debugging, trade-offs — from fundamentals to architect-level
Approach: Every topic starts with simple explanation → then interview-level depth
🧠 D → Driver (the boss who plans the work)
MEMORY MAP: SPARK ARCHITECTURE→DCE-SAT
DDriver (the boss who plans the work)
CCluster Manager (HR — allocates workers)
EExecutors (workers who do the actual processing)
SStages (assembly lines separated by shuffles)
AAQE (Adaptive Query Execution — auto-optimizer)
TTungsten (memory manager — off-heap, binary format)
SECTION 1: SPARK APPLICATION LIFECYCLE
Answer First: spark-submit starts the driver, which creates the SparkSession and builds a lazy DAG. An action creates a job; shuffle boundaries divide stages, tasks run on executors, and completion events return to the driver.
Memory Map: the complete lifecycle of a Spark application from spark-submit to job completion -> spark-submit launches the driver process -> driver builds a lazy DAG and jobs -> scheduler divides stages into executor tasks -> completion events close the application [01_Spark_Architecture_and_Internals.md:24].
Q1: Explain the complete lifecycle of a Spark application from spark-submit to job completion.
Simple Explanation:
Think of a Spark application as opening a restaurant kitchen for one big dinner service. Here is what happens step by step:
- The Head Chef (Driver) arrives and sets up the kitchen plan (SparkSession).
- The Head Chef calls the Restaurant Manager (Cluster Manager) and says "I need 10 line cooks tonight."
- The Restaurant Manager hires and assigns Line Cooks (Executors) to their stations.
- The Head Chef looks at the full dinner menu (your code) and plans the most efficient way to prepare all the dishes — this is the logical plan.
- A GPS navigation system (Catalyst Optimizer) finds the fastest route through the recipe — the physical plan.
- The Head Chef breaks the dinner into courses (Stages) — appetizer, then main, then dessert. Each course boundary is a "serving moment" (shuffle).
- Within each course, individual prep tasks are assigned — one per ingredient batch (one per partition).
- Tasks are handed to the line cooks, who execute them, report back, and the meal is served.
Technical Answer:
- The driver process starts and creates a
SparkContext/SparkSession
- SparkContext connects to the Cluster Manager (YARN/Mesos/K8s/Standalone)
- Cluster Manager allocates executor JVMs on worker nodes
- Driver converts user code into a logical plan (DAG of DataFrame operations)
- The Catalyst Optimizer optimizes the logical plan → physical plan
- The DAG Scheduler breaks the physical plan into stages at shuffle boundaries
- Each stage is broken into tasks (one per partition) by the Task Scheduler
- Tasks are serialized and sent to executors
- Executors run tasks, store results, and report back to the driver
- Results are collected or written to storage
# Example: a simple Spark application lifecycle in code
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("RestaurantKitchen") \ # ← Name your application (the dinner service)
.config("spark.executor.memory", "8g") \ # ← Each line cook gets 8 GB workspace
.getOrCreate() # ← Head Chef opens the kitchen (Driver starts)
df = spark.read.parquet("s3://orders/") # ← Read the ingredient list (lazy — no work yet!)
result = df.filter(df.status == "active") \ # ← Plan: only use fresh ingredients (still lazy)
.groupBy("region") \ # ← Plan: group dishes by region (will cause a shuffle = new stage)
.count() # ← Plan: count dishes per region (still lazy)
result.write.parquet("s3://output/") # ← ACTION! Now the kitchen actually starts cooking
# ^ This single action triggers: logical plan→Catalyst optimization → DAG → stages → tasks → execution
Interview Tip: They love to ask "walk me through what happens when you call .write()" — trace the full path from action to DAG to stages to tasks.
What NOT to Say: "Spark executes each line of code as you write it." No — Spark is lazy. Nothing happens until an action triggers the full pipeline.
Follow-up they'll ask: "What happens if the driver dies vs an executor dies?" → See Q5.
Answer First: The DAG Scheduler is the trip planner — it looks at your full itinerary and divides it into legs: "First drive from NYC to Philadelphia (Stage 1), then Philadelphia to Baltimore (Stage 2)." Each stop where you refuel is a shuffle boundary.
Memory Map: the difference between the DAG Scheduler and the Task Scheduler -> difference between dag scheduler and task scheduler distinguishes scheduler ownership before stage formation -> action creates a job DAG -> shuffle boundaries divide stages -> partitions become tasks -> scheduler events expose execution order [01_Spark_Architecture_and_Internals.md:76].
Q2: What is the difference between the DAG Scheduler and the Task Scheduler?
Simple Explanation:
Think of it like planning a road trip vs driving it.
The DAG Scheduler is the trip planner — it looks at your full itinerary and divides it into legs: "First drive from NYC to Philadelphia (Stage 1), then Philadelphia to Baltimore (Stage 2)." Each stop where you refuel is a shuffle boundary.
The Task Scheduler is the actual driver — for each leg of the trip, it decides: "Which lane should I be in? Should I take the highway or the side road?" It assigns individual driving tasks to available cars (executors), trying to pick the car closest to the data (data locality).
Technical Answer:
| Aspect | DAG Scheduler | Task Scheduler |
|---|
| Operates at | Stage level | Task level (within a stage) |
| Responsibility | Computes DAG of stages, identifies shuffle dependencies | Assigns tasks to executors with data locality |
| Handles | Stage retries on fetch failures | Individual task retries, speculative execution |
| Input | Logical plan / physical plan | TaskSet (set of tasks for one stage) |
| Locality | N/A | PROCESS_LOCAL > NODE_LOCAL > RACK_LOCAL > ANY |
Interview Tip: They may ask "How does Spark decide which executor runs which task?" — the answer is the Task Scheduler's locality-aware scheduling. It prefers the executor where data already lives.
What NOT to Say: "The DAG Scheduler assigns tasks to executors." No — the DAG Scheduler only creates stages. The Task Scheduler handles the actual assignment.
Answer First: Narrow dependency = cars staying in their own lane. Each car (partition) goes straight ahead without merging.
Memory Map: narrow vs wide dependencies with examples. Why does this distinction matter -> narrow vs wide dependencies with examples this distinction matter identifies the exact shuffle boundary and failure mode -> dependency determines redistribution -> map output crosses the exchange -> reducers fetch and sort partitions -> UI metrics reveal skew, spill, or failure [01_Spark_Architecture_and_Internals.md:101].
Q3: Explain narrow vs wide dependencies with examples. Why does this distinction matter?
Simple Explanation:
Imagine a highway.
Narrow dependency = cars staying in their own lane. Each car (partition) goes straight ahead without merging. No one needs to cross lanes. This is fast — no coordination needed. Examples: map, filter, union.
Wide dependency = a highway merge where cars from ALL lanes need to reorganize. Every car might need to move to a different lane based on its destination. This merge point is a shuffle — everyone has to slow down, signal, and reposition. Examples: groupByKey, join, repartition.
The merge is the most expensive part of the trip. Minimizing merges (shuffles) is the #1 Spark performance optimization.
Technical Answer:
df = spark.read.parquet("s3://sales/")
df2 = df.filter(df.amount > 100)
df3 = df2.withColumn("tax", df2.amount * 0.1)
df4 = df3.groupBy("region").sum("amount")
df4.write.parquet("s3://output/")
Why it matters: Wide dependencies trigger the most expensive operation in Spark — shuffle. Every shuffle means data serialization → disk write → network transfer → disk read → deserialization. Understanding this helps you minimize shuffles in your pipelines.
Interview Tip: They will ask "How do you reduce shuffles in a pipeline?" Answer: (1) filter early, (2) use broadcast joins, (3) co-partition data, (4) use reduceByKey instead of groupByKey.
What NOT to Say: "All joins cause shuffles." Not true — broadcast joins and co-partitioned joins avoid shuffles entirely.
Answer First: The Block Manager tracks cached and shuffle blocks on each executor, stores them in memory or disk, and serves local or remote block requests while reporting locations to the driver.
Memory Map: the Block Manager and how does it work -> executor registers blocks with its manager -> memory or disk stores partition replicas -> peer and driver requests locate cached data -> eviction and transfer metrics expose block state [01_Spark_Architecture_and_Internals.md:144].
Q4: What is the Block Manager and how does it work?
Simple Explanation:
Think of the Block Manager as a warehouse shelf system in each executor. Every executor has its own warehouse, and the warehouse stores:
- Cached data (items you need again soon — kept on the closest shelf)
- Shuffle data (items being shipped to other warehouses)
- Broadcast variables (company memos — one copy per warehouse)
There is also a central inventory tracker on the driver (BlockManagerMaster) that knows exactly which warehouse has which item. When Executor A needs a shuffle block from Executor B, it asks the central tracker "Where is block X?", gets the address, then fetches it directly.
Technical Answer:
BlockManager is the storage subsystem in each executor (and the driver). It manages:
- Cached/persisted RDD/DataFrame partitions
- Shuffle data (shuffle blocks)
- Broadcast variable blocks
- Task result blocks
Architecture:
- MemoryStore: On-heap and off-heap memory
- DiskStore: Local disk spillover
- BlockManagerMaster (on driver): Tracks all block locations across the cluster
- BlockTransferService (Netty-based): Fetches blocks from remote executors
When a shuffle reader needs a block from another executor, it queries BlockManagerMaster for the location, then uses BlockTransferService to fetch it.
Interview Tip: If asked about shuffle internals or caching, mention Block Manager — it shows you understand the storage layer beneath the abstractions.
What NOT to Say: "Cached data is stored on HDFS." No — cached data lives in executor memory/local disk, managed by Block Manager, not on distributed storage.
Answer First: An executor failure loses its in-memory partitions and active tasks, so Spark reschedules the work and recomputes missing data from lineage. A driver failure ends coordination and usually requires application restart, with streaming state recovered from checkpoints.
Memory Map: What happens when a driver fails vs when an executor fails -> driver loss removes scheduling authority -> executor loss invalidates local blocks and tasks -> cluster manager replaces available processes -> lineage and retry policy determine recovery [01_Spark_Architecture_and_Internals.md:175].
Q5: What happens when a driver fails vs when an executor fails?
Simple Explanation:
Back to our restaurant analogy:
If a line cook (executor) gets sick and goes home:
- The Head Chef notices (heartbeat timeout).
- The Head Chef reassigns that cook's dishes to other cooks.
- If that cook had already prepped something (shuffle output), the prep might need to be redone — unless you have a pantry system (External Shuffle Service) that kept the prep work safe.
If the Head Chef (driver) collapses:
- The entire kitchen shuts down. Nobody knows the plan anymore.
- In client mode: dinner service is cancelled. No recovery.
- In cluster mode: the Restaurant Manager can hire a new Head Chef, but the new chef has to start the plan from scratch — all in-progress work is lost.
Technical Answer:
Executor failure:
- Driver detects via heartbeat timeout
- Tasks on that executor are rescheduled on other executors
- If the stage used shuffle output from the lost executor, those shuffle blocks must be recomputed
- Cached RDD partitions on that executor are lost → recomputed on demand
- The External Shuffle Service mitigates this (shuffle data survives executor death)
Driver failure:
- Client mode: Entire application fails. No recovery.
- Cluster mode: With
spark.driver.supervise=true (Standalone) or YARN --max-app-attempts, the driver restarts, but ALL state is lost (SparkContext, accumulators, broadcast variables)
- Structured Streaming: Can recover from driver failure using checkpointing (offsets + state are persisted)
Interview Tip: Always mention the External Shuffle Service when discussing executor failure — it shows you know production-grade Spark.
What NOT to Say: "If the driver fails, executors continue working." No — executors cannot function without the driver. The driver is the brain.
SECTION 2: CATALYST OPTIMIZER
Answer First: Catalyst parses expressions, resolves tables and types, rewrites the logical plan, and compares physical strategies before Spark executes the selected plan.
Memory Map: the complete pipeline of Spark's Catalyst Optimizer -> complete pipeline of spark s catalyst optimizer selects the Catalyst phase that explains the observed plan -> expression enters parser -> analyzer resolves names and types -> optimizer rewrites the logical plan -> planner selects executable operators [01_Spark_Architecture_and_Internals.md:212].
Q6: Explain the complete pipeline of Spark's Catalyst Optimizer.
Simple Explanation:
Catalyst is like a GPS navigation system for your query. You type in your destination (SQL query or DataFrame code), and the GPS:
- Parses your input — "OK, you want to go from Home to Airport" (understands the request).
- Analyzes — "Let me verify 'Home' and 'Airport' are real places" (resolves table/column names).
- Optimizes the route — "Taking the highway is faster than side streets" (predicate pushdown, column pruning).
- Plans the physical drive — "Should I take Route A or Route B? Let me check traffic (statistics)" (chooses join strategies, picks best physical plan).
- Generates turn-by-turn directions — "Left in 200m, then merge right" (Tungsten code generation — optimized bytecode).
Just like GPS, the more information it has (traffic = table statistics), the better route it picks.
Technical Answer:
🧠 Memory Map
SQL/DataFrame API
↓
1. PARSING→Unresolved Logical Plan (AST)
↓
2. ANALYSIS→Resolved Logical Plan
(Analyzer resolves table names, column names, data types using Catalog)
↓
3. LOGICAL OPTIMIZATION→Optimized Logical Plan
Rule-based optimizations:
• Predicate pushdown
• Constant folding
• Column pruning (projection pushdown)
• Boolean simplification
• Null propagation
• Filter/projection combining
↓
4. PHYSICAL PLANNING→Physical Plan(s)
• Generates multiple candidate plans (e.g., SortMergeJoin vs BroadcastHashJoin)
• Cost model selects the best plan
• Uses table/column statistics if available (CBO)
↓
5. CODE GENERATION (Tungsten) → Optimized Java Bytecode
• Whole-stage code generation
• Fuses operators into tight loops
• Avoids virtual function dispatch
# See the Catalyst pipeline in action
df = spark.read.parquet("s3://sales/") # ← Read source data
result = df.filter(df.year == 2024) \ # ← Filter (Catalyst will push this down!)
.select("region", "amount") \ # ← Project (Catalyst will prune unused columns!)
.groupBy("region").sum("amount") # ← Aggregate
result.explain(True) # ← Show ALL 4 Catalyst phases
# Output shows:
# == Parsed Logical Plan == ← Step 1: raw AST
# == Analyzed Logical Plan == ← Step 2: columns/tables resolved
# == Optimized Logical Plan == ← Step 3: filter pushed down, columns pruned
# == Physical Plan == ← Step 4: HashAggregate chosen, codegen enabled
Key insight for interviews: Catalyst is why DataFrame operations are faster than RDD operations — the optimizer can reason about the operations and optimize the entire plan globally.
Interview Tip: If asked "Why are DataFrames faster than RDDs?", the answer is Catalyst + Tungsten. RDDs are opaque — Spark cannot optimize what it cannot see inside.
What NOT to Say: "Catalyst only works with SQL queries." No — it optimizes both SQL and DataFrame API calls identically. They both go through the same pipeline.
Answer First: Predicate pushdown = Instead of bringing ALL books to your desk and then searching, you tell the librarian "I only want books from 2024" and the librarian only brings you the 2024 shelf. The filter is pushed down to the source.
Memory Map: Predicate Pushdown and Projection Pushdown? When do they NOT work -> filter and column references reach the data source -> supported predicates and projections become scan requirements -> unsupported expressions remain Spark operators -> physical plan and scan metrics prove pushdown [01_Spark_Architecture_and_Internals.md:276].
Q7: What is Predicate Pushdown and Projection Pushdown? When do they NOT work?
Simple Explanation:
Imagine you are looking for a specific book in a library.
Predicate pushdown = Instead of bringing ALL books to your desk and then searching, you tell the librarian "I only want books from 2024" and the librarian only brings you the 2024 shelf. The filter is pushed down to the source.
Projection pushdown = Instead of photocopying the entire book, you say "I only need chapters 3 and 7." The librarian only copies those chapters. The column selection is pushed down to the source.
Both reduce the amount of data that ever enters Spark's processing pipeline.
Technical Answer:
- Predicate pushdown: Filters pushed as close to the data source as possible (into Parquet file metadata, JDBC WHERE clause). Reduces I/O dramatically.
- Projection pushdown: Only required columns are read from source. Columnar formats (Parquet) benefit hugely — entire column chunks are skipped.
When they DON'T work:
- UDFs in filter conditions — Catalyst cannot reason about UDF internals, so predicates involving UDFs cannot be pushed down
- Complex nested column access — may not be pushed in all Spark versions
- Non-optimized data sources that don't support pushdown
- After a shuffle — predicates before the shuffle cannot be pushed past it
df = spark.read.parquet("s3://sales/")
df.filter(df.year == 2024).select("region", "amount")
from pyspark.sql.functions import udf
is_valid = udf(lambda x: x > 0)
df.filter(is_valid(df.amount)).select("region", "amount")
How to verify: Use df.explain(True) — look for PushedFilters in the scan node.
Interview Tip: When discussing performance tuning, always mention checking explain() for pushed filters. If filters are not being pushed, it usually means a UDF is blocking optimization.
What NOT to Say: "Predicate pushdown always works automatically." It does not — UDFs, certain data sources, and complex expressions can prevent it.
Answer First: Adaptive Query Execution revises the physical plan from runtime statistics, so it can coalesce shuffle partitions, switch join strategies, and split skewed partitions after execution begins.
Memory Map: Adaptive Query Execution (AQE)? What problems does it solve -> adaptive query execution aqe problems it solve supplies the statistic that can revise physical execution -> statistics establish the initial strategy -> runtime observations revise joins or partitions -> reduced exchange and scan work -> final plan metrics prove adaptation [01_Spark_Architecture_and_Internals.md:320].
Q8: What is Adaptive Query Execution (AQE)? What problems does it solve?
Simple Explanation:
Remember our GPS analogy for Catalyst? Catalyst is like a GPS that plans your route before you start driving — based on estimated traffic. But what if there is an accident on the highway that the GPS did not know about?
AQE is a GPS that recalculates your route mid-drive based on actual traffic. It watches what is happening during execution and adjusts the plan on the fly. If a partition turns out to be tiny, AQE merges it. If a table turns out to be small, AQE switches to a broadcast join. If one lane is jammed (data skew), AQE splits the traffic.
Technical Answer:
AQE (default ON in Spark 3.x) re-optimizes the query plan at runtime based on actual shuffle statistics. It solves 3 problems:
1. Coalescing post-shuffle partitions:
- If many shuffle partitions are tiny, AQE merges them
- Config:
spark.sql.adaptive.coalescePartitions.enabled=true
- Eliminates the pain of tuning
spark.sql.shuffle.partitions
2. Converting Sort-Merge Join → Broadcast Hash Join:
- If one side of a join turns out to be small at runtime (<
spark.sql.adaptive.autoBroadcastJoinThreshold)
- Happens when compile-time statistics were wrong
3. Optimizing skew joins:
- Detects skewed partitions at runtime
- Splits the large partition and replicates the corresponding partition from the other side
- Config:
spark.sql.adaptive.skewJoin.enabled=true
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "2000")
Scenario question: "You set spark.sql.shuffle.partitions=200 but your 100 GB shuffle creates many tiny partitions. How does AQE help?" → AQE auto-coalesces the 200 partitions into fewer, larger partitions.
Interview Tip: AQE is THE modern answer to most Spark tuning questions. If asked "How do you tune shuffle partitions?", say "Set it high and let AQE coalesce. Manual tuning is legacy."
What NOT to Say: "I manually tune spark.sql.shuffle.partitions for each job." That is the pre-Spark 3.0 approach. AQE makes manual tuning largely unnecessary.
Answer First: Tungsten's whole-stage codegen collapses an entire stage of operators (filter → project → aggregate) into a single Java function.
Memory Map: Whole-Stage Code Generation (CodeGen) -> compatible operators form one generated pipeline -> Java source fuses row processing loops -> JVM compilation removes virtual-call overhead -> explain output marks whole-stage boundaries [01_Spark_Architecture_and_Internals.md:368].
Q9: What is Whole-Stage Code Generation (CodeGen)?
Simple Explanation:
Think of Tungsten's codegen like organizing your desk for maximum efficiency. Instead of reaching for a different tool for each step (open drawer, grab scissors, close drawer, open another drawer, grab tape...), you lay out everything you need in one line and process items in one smooth motion.
Without codegen, Spark processes data by calling one operator at a time — filter calls next() on scan, project calls next() on filter, etc. Each call has overhead (like opening/closing drawers). With codegen, Spark fuses all operators into a single tight loop — one function that does scan+filter+project in one pass. No overhead between steps.
Technical Answer:
Tungsten's whole-stage codegen collapses an entire stage of operators (filter → project → aggregate) into a single Java function.
Traditional Volcano/iterator model:
- Each operator calls
next() on its child
- Virtual function dispatch per row
- Poor CPU cache utilization
- Branch prediction misses
With CodeGen:
- Fuses operators into tight loops
- No virtual function calls
- Operates on raw memory (
sun.misc.Unsafe)
- Leverages CPU pipelining and L1/L2 cache
- Can see generated code:
df.queryExecution.debug.codegen()
df = spark.read.parquet("s3://sales/")
result = df.filter(df.amount > 100).select("region", "amount").groupBy("region").sum("amount")
result.explain()
What breaks codegen:
- External sorts
- Some joins with complex expressions
- Python UDFs (completely bypass codegen)
- Very large expressions (hit JVM method size limit)
Interview Tip: If asked "Why are Python UDFs slow?", mention that they bypass Tungsten codegen entirely. Suggest Pandas UDFs (Arrow-based) as the alternative.
What NOT to Say: "CodeGen makes all operations faster." It does not help I/O-bound operations, and Python UDFs bypass it completely.
Answer First: If Catalyst's rule-based optimizer is a GPS that follows fixed rules ("always prefer highways"), then CBO is like the GPS checking real-time traffic data before choosing a route.
Memory Map: Cost-Based Optimization (CBO) -> catalog table and column statistics estimate cardinality -> cardinality model compares candidate physical operators -> planner selects the lowest estimated plan -> EXPLAIN output exposes estimates and choice [01_Spark_Architecture_and_Internals.md:420].
Q10: What is Cost-Based Optimization (CBO)?
Simple Explanation:
If Catalyst's rule-based optimizer is a GPS that follows fixed rules ("always prefer highways"), then CBO is like the GPS checking real-time traffic data before choosing a route.
CBO uses statistics about your data — how many rows, how many distinct values, min/max values — to make smarter decisions. Without statistics, Spark guesses. With statistics, Spark knows.
Technical Answer:
CBO uses table and column statistics to choose optimal plans.
Collect statistics:
ANALYZE TABLE t COMPUTE STATISTICS;
ANALYZE TABLE t COMPUTE STATISTICS FOR COLUMNS c1, c2;
Statistics collected:
- Table: row count, size in bytes
- Column: distinct count, min, max, avg length, null count, histogram
What CBO affects:
- Join strategy selection (broadcast vs sort-merge)
- Join ordering in multi-table joins
- Filter selectivity estimation
Enable:
spark.conf.set("spark.sql.cbo.enabled", "true")
spark.conf.set("spark.sql.cbo.joinReorder.enabled", "true")
Interview Tip: If asked "Your broadcast join threshold is 10 MB, but Spark is still doing sort-merge join on a 5 MB table. Why?", the answer is: Spark does not know the table is 5 MB because statistics have not been collected. Run ANALYZE TABLE.
What NOT to Say: "CBO is always on by default and works automatically." You need to explicitly collect statistics and enable it.
SECTION 3: MEMORY MANAGEMENT
Answer First: Reserved corner (300 MB) — always occupied by essentials (pens, stapler). You cannot use this for work.
Memory Map: Spark's Unified Memory Management model in detail -> spark s unified memory management model in detail locates pressure in the relevant Spark memory pool -> partition demand consumes execution memory -> cached blocks compete for the shared region -> spill or garbage collection signals pressure -> executor metrics locate the constrained pool [01_Spark_Architecture_and_Internals.md:468].
Q11: Explain Spark's Unified Memory Management model in detail.
Simple Explanation:
Think of each executor's memory as an office desk. The desk has fixed zones:
- Reserved corner (300 MB) — always occupied by essentials (pens, stapler). You cannot use this for work.
- Personal area (User Memory, 40%) — your personal stuff: sticky notes, coffee mug (UDF variables, RDD metadata).
- Work area (Spark Unified Memory, 60%) — the actual work surface, split into two halves:
- Active project zone (Execution) — papers you are actively working on right now (shuffles, joins, sorts)
- Reference shelf (Storage) — documents you might need again soon (cached DataFrames, broadcast variables)
The key insight: the boundary between Execution and Storage is flexible. If you are doing a massive join and need more desk space, Execution can push Storage papers aside. But Storage CANNOT push active work aside — you cannot pause mid-calculation.
Technical Answer:
📐 Architecture Diagram
Executor JVM Heap
┌────────────────────────────────────────────┐
│ Reserved Memory (300 MB, fixed) │
├────────────────────────────────────────────┤
│ User Memory │
│ (1 - spark.memory.fraction) * (heap-300 MB) │
│ Default: 40% of (heap - 300 MB) │
│ Used for: UDF variables, RDD metadata │
├────────────────────────────────────────────┤
│ Spark (Unified) Memory │
│ spark.memory.fraction * (heap - 300 MB) │
│ Default: 60% of (heap - 300 MB) │
│ ┌──────────────┬─────────────────┐ │
│ │ Execution │ Storage │ │
│ │ (shuffles, │ (cached │ │
│ │ joins, │ data, │ │
│ │ sorts) │ broadcast) │ │
│ │ │ │ │
│ │ ← soft boundary, can borrow → │ │
│ └──────────────┴─────────────────┘ │
└────────────────────────────────────────────┘
Key rule: Execution can evict Storage (cached data), but Storage cannot evict Execution. This is because execution memory is critical (can't pause mid-computation), while cached data can be recomputed.
Interview Tip: Draw this diagram on the whiteboard. Interviewers love it. Emphasize the "soft boundary" and why Execution wins over Storage.
What NOT to Say: "Execution and Storage memory are fixed, separate pools." That was the OLD model (Static Memory Management, pre-Spark 1.6). The Unified model has a flexible boundary.
Answer First: Executor memory = the main office room (JVM heap)
Memory Map: all memory-related configurations and their relationship -> executor heap divides reserved and unified regions -> storage and execution borrow shared memory -> overhead controls native and Python processes -> spill and garbage collection reveal sizing errors [01_Spark_Architecture_and_Internals.md:524].
Simple Explanation:
Think of it like renting an office space. You need to understand what you are paying for:
- Executor memory = the main office room (JVM heap)
- Memory overhead = the hallway, lobby, and bathrooms (off-heap: VM internals, network buffers)
- Container size = total rent = office room + hallway (YARN allocates the whole thing)
If you only configure the office room but forget the hallway, YARN kills your container for using more space than you paid for.
Technical Answer:
spark.executor.memory = "8g"
spark.executor.memoryOverhead = "2g"
spark.memory.fraction = 0.6
spark.memory.storageFraction = 0.5
spark.memory.offHeap.enabled = true
spark.memory.offHeap.size = "4g"
spark.driver.memory = "4g"
spark.driver.maxResultSize = "1g"
Interview Tip: The most common production issue is YARN killing containers. Always mention memoryOverhead — it is the most overlooked config.
What NOT to Say: "Just increase spark.executor.memory to fix OOM errors." Sometimes the issue is overhead, sometimes it is skew. Diagnose first.
Answer First: OOM (Out of Memory) errors are the 1 reason Spark jobs fail in production. There are two kinds.
Memory Map: What causes OOM errors? How do you debug driver OOM vs executor OOM -> driver heap pressure affects planning or result collection -> executor pressure occurs inside task memory -> logs and GC metrics locate the failing process -> sizing or partition repair targets that pool [01_Spark_Architecture_and_Internals.md:562].
Q13: What causes OOM errors? How do you debug driver OOM vs executor OOM?
Simple Explanation:
OOM (Out of Memory) errors are the #1 reason Spark jobs fail in production. There are two kinds:
Driver OOM = The Head Chef's desk overflows. Usually because someone called collect() and tried to bring the entire warehouse to the chef's tiny desk. Or a broadcast variable is too large to fit in the driver's memory.
Executor OOM = A line cook's station overflows. Usually because one cook got an unfairly large portion of data (skew), or there is not enough memory for a massive shuffle/join.
Technical Answer:
Driver OOM causes:
| Cause | Solution |
|---|
collect() on large data | Use take(n), show(), or write to storage |
| Large broadcast variable | Check variable size, increase spark.driver.memory |
| Too many tasks (metadata) | Reduce number of partitions |
| Accumulator results | Limit accumulator usage |
toPandas() on large DF | Use Arrow + batch conversion |
Executor OOM causes:
| Cause | Solution |
|---|
| Skewed partition (1 partition >> others) | Salt keys, use AQE, broadcast join |
| Insufficient memory for shuffle/join | Increase spark.executor.memory, increase partitions |
| UDF holding large objects | Move large objects to broadcast variables |
| Container killed by YARN | Increase spark.executor.memoryOverhead |
Debugging checklist:
Interview Tip: They will give you a scenario: "Your job ran for 6 hours and then failed with OOM. Walk me through debugging." Follow the 4-step checklist above.
What NOT to Say: "I would just double the executor memory." Blindly increasing memory without diagnosing the root cause is wasteful and often does not fix skew-related OOM.
Answer First: Tungsten is like organizing your desk for maximum efficiency — using a binary layout with no wasted space.
Memory Map: Tungsten's memory management? What is sun.misc.Unsafe -> unsafe binary rows avoid JVM object overhead -> off-heap pages reduce garbage collection -> generated operators process compact memory directly -> spill and CPU metrics expose the tradeoff [01_Spark_Architecture_and_Internals.md:615].
Q14: What is Tungsten's memory management? What is sun.misc.Unsafe?
Simple Explanation:
Tungsten is like organizing your desk for maximum efficiency — using a binary layout with no wasted space.
Normally, Java stores objects with a lot of overhead: object headers, pointers, alignment padding. It is like storing each document in its own fancy folder with labels and dividers — wastes a lot of space.
Tungsten says: "Forget the fancy folders. Let me lay out all the data as raw bytes in a flat row on the desk." No wasted space, no folders to open and close. And since Tungsten manages memory directly (bypassing the JVM garbage collector), there are no "cleaning crew interruptions" (GC pauses).
sun.misc.Unsafe is the low-level Java API that gives Tungsten direct access to memory addresses — like having a master key to every shelf and drawer, bypassing all the normal Java safety locks.
Technical Answer:
Tungsten manages memory outside the JVM garbage collector using sun.misc.Unsafe:
- Direct memory allocation/deallocation (like C
malloc/free)
- Read/write raw bytes at memory addresses
- No GC overhead for managed data
Tungsten stores data in a compact binary format: rows are serialized into byte arrays with:
- Null bitmap
- Fixed-length values (int, long, double)
- Variable-length region (strings, arrays)
📐 Architecture Diagram
# Tungsten binary row layout (conceptual):
# ┌──────────┬─────────┬─────────┬──────────────────┐
# │ Null bits │ int(4B) │ long(8B)│ string (offset+len)│
# └──────────┴─────────┴─────────┴──────────────────┘
# No object headers, no pointers, no padding waste
# Everything packed tight → better CPU cache utilization
Benefits: No GC pauses, better cache locality, explicit memory management, smaller memory footprint.
Interview Tip: When discussing Spark performance, mention Tungsten as the reason DataFrames are fast at the memory level (Catalyst optimizes the plan, Tungsten optimizes the execution).
What NOT to Say: "Tungsten only works with off-heap memory." Tungsten's binary format works with both on-heap and off-heap memory. Off-heap is optional.
SECTION 4: SHUFFLE DEEP DIVE
Answer First: A shuffle is like a postal sorting system. Imagine you have 100 post offices (map tasks) and each office has letters for 50 different cities (reduce partitions).
Memory Map: the complete shuffle process in Spark -> map tasks partition records into reducer buckets -> local files retain each bucket -> downstream tasks fetch and merge remote blocks -> exchange metrics expose bytes spill and skew [01_Spark_Architecture_and_Internals.md:656].
Q15: Explain the complete shuffle process in Spark.
Simple Explanation:
A shuffle is like a postal sorting system. Imagine you have 100 post offices (map tasks) and each office has letters for 50 different cities (reduce partitions).
Map side (Shuffle Write):
Each post office sorts all its letters by destination city, bundles them, and puts the bundles in outgoing mailboxes (writes to disk).
Reduce side (Shuffle Read):
Each destination city sends a truck to ALL 100 post offices to collect its bundle. The truck drives around, picks up all the bundles, and brings them home for final delivery (processing).
This is why shuffle is expensive — every city needs to visit every post office. It is an all-to-all data exchange.
Technical Answer:
Map side (Shuffle Write):
- Each map task computes which reducer partition each record belongs to (hash or range partitioner)
- Records written to
SortShuffleWriter (default since Spark 2.0)
- Data sorted by partition ID → written to a single data file + index file per map task
- Number of shuffle files = number of map tasks (NOT map x reduce)
Reduce side (Shuffle Read):
- Each reduce task fetches its partition from ALL map tasks via
BlockStoreShuffleReader
- Uses
ShuffleClient (Netty) for remote block fetching
- Data deserialized and optionally sorted (for sort-merge operations)
df = spark.read.parquet("s3://sales/")
grouped = df.groupBy("region").sum("amount")
grouped.write.parquet("s3://output/")
Why shuffle is expensive:
- Disk I/O (write on map side, read on reduce side)
- Network I/O (cross-executor data transfer)
- Serialization/deserialization overhead
- Can cause spill to disk if memory insufficient
Interview Tip: If asked "What is the most expensive operation in Spark?", the answer is always shuffle. Follow up with how to minimize it (filter early, broadcast joins, co-partitioning).
What NOT to Say: "Shuffle files = map tasks x reduce tasks." That was the old Hash Shuffle Manager. The modern Sort Shuffle Manager creates only one file per map task.
Answer First: This config tells Spark: "After a shuffle, split the data into this many pieces." The default is 200, which is almost never right.
Memory Map: spark.sql.shuffle.partitions and how do you tune it -> post-exchange data size determines reducer demand -> target bytes per task estimate a useful count -> available cores bound effective parallelism -> task duration and spill refine the setting [01_Spark_Architecture_and_Internals.md:708].
Q16: What is spark.sql.shuffle.partitions and how do you tune it?
Simple Explanation:
This config tells Spark: "After a shuffle, split the data into this many pieces." The default is 200, which is almost never right.
Think of it like a pizza. If you cut a small pizza into 200 slices, each slice is uselessly tiny. If you cut a giant pizza into only 200 slices, each slice is too big to eat. You need the right number of slices for the pizza size.
Technical Answer:
- Default: 200 (often too low for large data or too high for small data)
- This controls the number of partitions after a shuffle (groupBy, join, repartition by column)
Tuning formula:
num_partitions = total_shuffle_data_size / target_partition_size
target_partition_size = 128 MB to 200 MB
Example: 50 GB shuffle data → 50 GB / 200 MB = 250 partitions
spark.conf.set("spark.sql.shuffle.partitions", "250")
spark.conf.set("spark.sql.shuffle.partitions", "2000")
spark.conf.set("spark.sql.adaptive.enabled", "true")
With AQE: Set this high (e.g., 2000) and let coalescePartitions merge small partitions automatically. This is the modern best practice.
Common mistake: Setting it to 200 for both a 1 GB dataset and a 1 TB dataset.
Interview Tip: Always mention AQE as the modern approach. Manual tuning is a red flag that you are stuck in pre-Spark 3.0 thinking.
What NOT to Say: "I always use the default 200." That one answer tells the interviewer you have never tuned Spark at scale.
Answer First: If a cook goes home sick, all the prepped ingredients on their station are lost — the next stage needs to redo the prep.
Memory Map: the External Shuffle Service and why is it critical -> auxiliary service preserves map outputs beyond executor lifetime -> dynamic allocation can remove idle workers safely -> later reducers still fetch registered blocks -> lost-output errors reveal missing service support [01_Spark_Architecture_and_Internals.md:747].
Q17: What is the External Shuffle Service and why is it critical?
Simple Explanation:
Imagine your line cooks (executors) prep ingredients (shuffle data) and leave them on their stations. If a cook goes home sick, all the prepped ingredients on their station are lost — the next stage needs to redo the prep.
The External Shuffle Service is like a shared pantry on each node. Cooks put their prepped ingredients in the pantry, not on their personal station. If a cook leaves, the ingredients are still in the pantry for anyone to use.
This is critical for dynamic allocation — Spark can scale executors up and down without losing shuffle data.
Technical Answer:
A long-running auxiliary service on each worker node that serves shuffle files independently of executors.
Why critical:
- Dynamic allocation: Executors can be removed (scaled down) without losing their shuffle files. Without it, removing an executor means recomputing its shuffle output.
- Fault tolerance: If executor crashes, shuffle data still available
- Resource efficiency: Executors released between stages while shuffle output remains accessible
Config:
spark.conf.set("spark.shuffle.service.enabled", "true")
spark.conf.set("spark.dynamicAllocation.enabled", "true")
Interview Tip: If asked about dynamic allocation, ALWAYS mention External Shuffle Service as a prerequisite. They are tightly coupled.
What NOT to Say: "Dynamic allocation just adds and removes executors automatically." You must mention that without ESS, removing executors loses shuffle data.
Answer First: FetchFailedException means: "A reduce task tried to pick up its package from another executor, but the package was not there." It is like a delivery truck arriving at a warehouse and finding it closed.
Memory Map: you handle the "Fetch Failed" exception -> reducer reports an unavailable map block -> scheduler invalidates the missing output -> upstream partition reruns on a healthy worker -> repeated failures point to network disk or executor loss [01_Spark_Architecture_and_Internals.md:777].
Q18: How do you handle the "Fetch Failed" exception?
Simple Explanation:
FetchFailedException means: "A reduce task tried to pick up its package from another executor, but the package was not there." It is like a delivery truck arriving at a warehouse and finding it closed.
This is one of the most common production errors. The package (shuffle data) is missing because the executor that produced it either crashed, ran out of memory, or had a network issue.
Technical Answer:
FetchFailedException = reduce task couldn't fetch shuffle data from a map task's executor.
Causes:
- Executor OOM/crash during or after shuffle write
- Network issues (timeout, connection refused)
- Disk failures
- Long GC pauses making executor unresponsive
Default behavior: Spark retries the entire stage (spark.stage.maxConsecutiveAttempts)
Fixes:
spark.conf.set("spark.shuffle.service.enabled", "true")
spark.conf.set("spark.executor.memory", "16g")
spark.conf.set("spark.shuffle.io.maxRetries", "10")
spark.conf.set("spark.shuffle.io.retryWait", "10s")
Interview Tip: This is a scenario question favorite: "Your job fails with FetchFailedException at 90% completion. What do you do?" Follow the 5 fixes above in order.
What NOT to Say: "Just retry the job." The retry will hit the same issue. You need to fix the root cause.
Answer First: Shuffle spill is like your desk overflowing during a big project. You run out of desk space (memory), so you start putting papers on the floor (disk).
Memory Map: shuffle spill. How do you detect and minimize it -> sort buffers exceed execution memory -> records flush to temporary disk runs -> merge work adds I/O and serialization -> memory and disk spill counters quantify pressure [01_Spark_Architecture_and_Internals.md:821].
Q19: Explain shuffle spill. How do you detect and minimize it?
Simple Explanation:
Shuffle spill is like your desk overflowing during a big project. You run out of desk space (memory), so you start putting papers on the floor (disk). Working from the floor is much slower than working from your desk — you have to bend down, pick up papers, bring them back to the desk.
In Spark terms: when execution memory is full during a shuffle, data "spills" from memory to local disk. Disk is 10-100x slower than memory.
Technical Answer:
Shuffle spill occurs when execution memory is exhausted during shuffle operations. Data spills from memory to disk.
Detect in Spark UI:
- Stage detail page → Spill (Memory) and Spill (Disk) columns
- Spill (Memory) > 0 indicates memory pressure
Minimize:
# 1. Give executors more memory
spark.conf.set("spark.executor.memory", "16g")
# 2. Increase Spark's share of the heap
spark.conf.set("spark.memory.fraction", "0.7") # ← Default 0.6, increase to 0.7
# 3. Increase parallelism (smaller data per task = less memory needed per task)
spark.conf.set("spark.sql.shuffle.partitions", "1000") # ← More partitions = smaller chunks
# 4. Filter early and select only needed columns BEFORE the shuffle
df = df.filter(df.year == 2024) \ # ← Reduce data volume before shuffle
.select("region", "amount") \ # ← Drop unnecessary columns
.groupBy("region").sum("amount") # ← Now the shuffle is much smaller
# 5. Use mapPartitions instead of map (reduces per-row object overhead)
# 6. Consider off-heap memory (avoids GC pressure on the heap)
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "4g")
Interview Tip: When looking at Spark UI, always check Spill columns. If Spill (Disk) is large, you have a memory problem that is silently slowing your job.
What NOT to Say: "Spill is normal and acceptable." Small spills are OK, but large spills (GBs) indicate a serious performance problem. A job with heavy spill can be 10x slower.
SECTION 5: SERIALIZATION
Answer First: Serialization is how Spark "packs" data for shipping between executors. Think of it like packing for a move.
Memory Map: all Spark serialization options with trade-offs -> record shape selects an encoding -> encoding changes payload size and CPU -> transfer crosses process or network boundaries -> task duration exposes the trade-off [01_Spark_Architecture_and_Internals.md:865].
Q20: Compare all Spark serialization options with trade-offs.
Simple Explanation:
Serialization is how Spark "packs" data for shipping between executors. Think of it like packing for a move:
- Java serialization = throwing everything into garbage bags. Works for anything, but bulky and slow to unpack.
- Kryo = using labeled moving boxes. 10x faster, more compact, but you need to label each box type (register classes).
- Tungsten binary = vacuum-sealing everything flat. Most compact, fastest. Used automatically by DataFrames.
- Apache Arrow = using standardized shipping containers. Great for cross-system transfer (PySpark to Pandas).
Technical Answer:
| Serialization | Speed | Size | Use Case |
|---|
| Java (default for RDD) | Slow | Large | Any Serializable class. Avoid for performance. |
| Kryo | 10x faster | Compact | RDD operations. Must register classes. |
| Tungsten binary | Fastest | Most compact | DataFrames/Datasets internally. Not configurable. Off-heap, no GC. |
| Apache Arrow | Very fast | Columnar | PySpark ↔ Pandas conversion. Zero-copy. Enable with spark.sql.execution.arrow.pyspark.enabled=true |
Kryo config:
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
spark.conf.set("spark.kryo.registrationRequired", "true")
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
pdf = df.toPandas()
Key interview point: DataFrames are always faster than RDDs because they use Tungsten binary format internally, bypassing Java/Kryo serialization entirely.
Interview Tip: If asked "How do you speed up PySpark toPandas()?", the answer is Arrow. If asked "How do you speed up RDD operations?", the answer is Kryo.
What NOT to Say: "I use Java serialization because it is the default." Java serialization is the slowest option. Always switch to Kryo for RDDs.
Answer First: PySpark is like a bilingual translator between Python and Java. Your Python code does not run directly on the cluster. Instead.
Memory Map: PySpark actually execute Python code? Explain the Py4J gateway architecture -> pyspark actually execute python code py4j gateway architecture defines how data crosses the Python and JVM boundary -> API call builds a JVM plan -> Python callback crosses a process boundary -> rows or Arrow batches execute externally -> worker time quantifies the penalty [01_Spark_Architecture_and_Internals.md:905].
Q21: How does PySpark actually execute Python code? Explain the Py4J gateway architecture.
Simple Explanation:
PySpark is like a bilingual translator between Python and Java. Your Python code does not run directly on the cluster. Instead:
- You write Python code.
- A translator (Py4J) converts your Python calls into Java calls on the Driver.
- The Java Driver sends work to Java Executors — no Python on the executors for native DataFrame operations.
- BUT if you use a Python UDF, each executor has to spawn a Python worker process, ship data to it via a socket, wait for it to process, and get results back. This Python worker is the bottleneck.
Think of it like a meeting with an interpreter. If everyone speaks the same language (Java/DataFrame ops), communication is instant. But if someone insists on speaking a different language (Python UDF), everything has to be translated back and forth — slow.
Technical Answer:
📐 Architecture Diagram
┌─────────────────────┐ ┌─────────────────────┐
│ Python Driver │ │ JVM Driver │
│ │ Py4J │ │
│ PySpark API calls ─┼──────┼→ Java SparkContext │
│ │ │ │
└─────────────────────┘ └──────────┬───────────┘
│
┌─────────┴──────────┐
│ Executors (JVM) │
│ ┌─────────────────┐│
│ │ Python Worker ││ ← Spawned for UDFs
│ │ (subprocess) ││
│ └─────────────────┘│
└─────────────────────┘
How it works:
- Python
SparkSession wraps a Java SparkSession via Py4J gateway
- DataFrame operations are translated to JVM calls → no Python on executors for native operations
- When a Python UDF is used:
- Each executor spawns a Python worker process
- Data serialized (pickle or Arrow) → sent via socket to Python worker
- Python worker processes data → sends results back
- This JVM↔Python boundary is the main performance overhead of PySpark vs Scala Spark
df.filter(df.amount > 100).groupBy("region").sum("amount")
from pyspark.sql.functions import udf
@udf("double")
def add_tax(amount):
return amount * 1.1
df.withColumn("total", add_tax(df.amount))
from pyspark.sql.functions import pandas_udf
@pandas_udf("double")
def add_tax_fast(amount: pd.Series) -> pd.Series:
return amount * 1.1
df.withColumn("total", add_tax_fast(df.amount))
Key insight: Pure DataFrame/SQL operations in PySpark are just as fast as Scala because they run entirely in the JVM. The overhead only appears with Python UDFs.
Interview Tip: If asked "Is PySpark slower than Scala Spark?", the nuanced answer is: "For DataFrame/SQL operations, they are identical. The difference appears only with Python UDFs. Use Pandas UDFs to minimize the gap."
What NOT to Say: "PySpark is always slower than Scala." This is a common misconception. For native operations (99% of pipeline code), there is zero performance difference.
SECTION 6: FAULT TOLERANCE & SPECULATION
Answer First: Spark's fault tolerance is based on a simple but powerful idea: remember how to redo the work, not the work itself.
Memory Map: Spark achieve fault tolerance -> spark achieve fault tolerance selects lineage replay or speculative retry semantics -> lost or slow task triggers recovery -> lineage replay or duplicate attempt recomputes output -> idempotent winner is retained -> retry metrics confirm recovery [01_Spark_Architecture_and_Internals.md:977].
Q22: How does Spark achieve fault tolerance?
Simple Explanation:
Spark's fault tolerance is based on a simple but powerful idea: remember how to redo the work, not the work itself.
Instead of replicating data across nodes (like HDFS does), Spark keeps a "recipe" (lineage) for every piece of data. If data is lost (executor crashes), Spark reruns the recipe to recreate it. This is like a chef who does not save every dish in a freezer, but instead keeps the recipe book — if a dish is dropped, they just cook it again.
Technical Answer:
- RDD lineage: Each RDD knows how to reconstruct itself from its parent. If a partition is lost, Spark replays the transformations to recreate it.
- Checkpointing: Breaks lineage by saving data to reliable storage. Two types:
- Reliable checkpoint: Saves to HDFS/S3 (fault-tolerant)
- Local checkpoint: Saves to executor local storage (not fault-tolerant, faster)
- Stage retry: If a fetch failure occurs, the entire stage is recomputed
- Task retry: Individual tasks are retried on failure (default: 4 attempts,
spark.task.maxFailures)
- Structured Streaming: Checkpoints track offsets + state for exactly-once recovery
spark.sparkContext.setCheckpointDir("s3://checkpoints/")
df = spark.read.parquet("s3://data/")
for i in range(100):
df = df.withColumn("score", some_transform(df.score))
if i % 10 == 0:
df = df.checkpoint()
df.count()
Interview Tip: If asked "How does Spark handle failures?", start with lineage (the core idea), then mention checkpointing as the optimization for long lineages.
What NOT to Say: "Spark replicates data like HDFS for fault tolerance." No — Spark uses lineage (re-computation), not replication. That is the fundamental design difference.
Answer First: Whichever delivers first, you keep that order and cancel the other. You pay a little extra (resources), but you are guaranteed to get your food fast.
Memory Map: speculative execution? When should you enable vs disable it -> scheduler detects an abnormally slow attempt -> duplicate work launches on another worker -> first successful result wins -> skew and side effects can make duplication harmful [01_Spark_Architecture_and_Internals.md:1015].
Q23: What is speculative execution? When should you enable vs disable it?
Simple Explanation:
Imagine you order food from two delivery services simultaneously. Whichever delivers first, you keep that order and cancel the other. You pay a little extra (resources), but you are guaranteed to get your food fast.
Speculative execution does the same thing: if one task is running much slower than others (a "straggler"), Spark launches a copy of that task on another executor. Whichever copy finishes first wins. The straggler is killed.
Technical Answer:
Spark re-launches slow tasks ("stragglers") on other executors and takes the result from whichever finishes first.
Config:
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "3")
spark.conf.set("spark.speculation.quantile", "0.75")
Enable when:
- Tasks have variable duration due to hardware issues or data locality
- You're running on heterogeneous hardware
- Network storage has variable latency
Disable when:
- Tasks have side effects (non-idempotent writes) — speculation would write duplicates
- Slowness is due to data skew — the re-launched task processes the same skewed partition
- You're already using most of cluster resources
Interview Tip: Always pair this with the skew caveat. "Speculation does NOT fix skew — if the task is slow because of a huge partition, the duplicate task will be equally slow."
What NOT to Say: "Speculation fixes slow tasks." It fixes stragglers caused by hardware/network issues, NOT tasks that are slow due to data skew.
Answer First: Broadcast variable = The Head Chef posts the daily specials menu on the kitchen wall. Every cook can read it, but nobody can change it.
Memory Map: Accumulators and Broadcast variables? What are the pitfalls -> driver publishes each broadcast value -> executor caches a read-only copy -> accumulator updates may repeat during task retry -> driver observes merged diagnostic values [01_Spark_Architecture_and_Internals.md:1048].
Q24: What are Accumulators and Broadcast variables? What are the pitfalls?
Simple Explanation:
In the restaurant kitchen:
Broadcast variable = The Head Chef posts the daily specials menu on the kitchen wall. Every cook can read it, but nobody can change it. It is a read-only shared reference. Instead of giving every cook their own copy (shipping with every task), one copy is posted per kitchen (executor).
Accumulator = A click counter at the door. Every time a customer walks in, the counter goes up. Cooks can add to it, but only the Head Chef can read the total. It is a write-only (from executor perspective) shared counter.
Technical Answer:
Broadcast variables:
- Read-only variables cached on each executor (not shipped with every task)
- Use for large lookup tables
lookup = spark.sparkContext.broadcast(large_dict)
df.filter(col("key").isin(lookup.value.keys()))
- Pitfalls:
- If too large → OOM on driver (collects before broadcasting)
- Memory used =
table_size x num_executors
- Must
.unpersist() or .destroy() manually
Accumulators:
- Write-only variables "added" to by executors, readable only by driver
counter = spark.sparkContext.accumulator(0)
rdd.foreach(lambda x: counter.add(1))
print(counter.value)
- Critical pitfall: In transformations (not actions), accumulators may be incremented more than once if tasks are retried or stages re-executed. Only use accumulators inside actions for guaranteed exactly-once semantics.
counter = spark.sparkContext.accumulator(0)
rdd2 = rdd.map(lambda x: (counter.add(1), x)[1])
rdd2.count()
counter = spark.sparkContext.accumulator(0)
rdd.foreach(lambda x: counter.add(1))
print(counter.value)
Interview Tip: The accumulator pitfall is a classic gotcha question. Always mention: "Accumulators are only guaranteed accurate inside actions, not transformations."
What NOT to Say: "Accumulators are like global variables you can read and write from anywhere." They are write-only from executors and only accurately reflect counts when used inside actions.
SECTION 7: SCENARIO-BASED ARCHITECTURE QUESTIONS
Answer First: The host (Task Scheduler) seats the first 100 customers. As each table finishes (task completes), the next customer in line is seated.
Memory Map: Scenario — Your Spark job has 1000 tasks but only 100 executor cores. How does Spark schedule them -> available cores admit one task per configured CPU share -> surplus tasks wait in scheduler queues -> completed waves release slots for successors -> timeline confirms ten scheduling waves [01_Spark_Architecture_and_Internals.md:1101].
Q25: Scenario — Your Spark job has 1000 tasks but only 100 executor cores. How does Spark schedule them?
Simple Explanation:
Think of a restaurant with 100 tables but 1000 customers waiting. You cannot seat everyone at once. The host (Task Scheduler) seats the first 100 customers. As each table finishes (task completes), the next customer in line is seated. With 1000 customers and 100 tables, you need roughly 10 waves to serve everyone.
Spark adds a locality preference: it tries to seat each customer at the table closest to the kitchen (data). If their preferred table is not available within 3 seconds (spark.locality.wait), they get seated anywhere.
Technical Answer:
- Spark doesn't run all 1000 tasks simultaneously
- The Task Scheduler maintains a queue of pending tasks
- It assigns tasks to available slots (one slot = one core) respecting data locality
- When a task completes, the next pending task is scheduled on the freed slot
- If
spark.locality.wait (default 3s) expires and no local slot is available, the task is scheduled on a less-local slot
- Throughput: ~100 tasks running at any time, with 10 waves of tasks
Interview Tip: Follow up with: "If all 10 waves take similar time, your parallelism is good. If the last wave has only 5 tasks while 95 cores sit idle, you have a tail problem — increase partitions."
What NOT to Say: "Spark waits until all 1000 tasks are ready before starting." No — Spark starts immediately with whatever cores are available and processes in waves.
Answer First: This is like using a forklift to move a shoebox. The 50 MB table is tiny — it should be broadcast (copied) to every executor so the join happens locally without any shuffle. But Spark is using a Sort-Merge Join (the forklift), which shuffles both tables across the network.
Memory Map: Scenario — You're running a join between a 500 GB and a 50 MB table. The job is doing a Sort-Merge Join. What's wrong -> re running join between 500 gb and 50 mb table job doing sort merge join s determines the join operator and required data movement -> input sizes and key distribution shape the join -> broadcast or exchange moves data -> physical operator combines rows -> shuffle and skew metrics validate selection [01_Spark_Architecture_and_Internals.md:1122].
Q26: Scenario — You're running a join between a 500 GB and a 50 MB table. The job is doing a Sort-Merge Join. What's wrong?
Simple Explanation:
This is like using a forklift to move a shoebox. The 50 MB table is tiny — it should be broadcast (copied) to every executor so the join happens locally without any shuffle. But Spark is using a Sort-Merge Join (the forklift), which shuffles both tables across the network.
The GPS (Catalyst) chose the wrong route because it does not know the table is only 50 MB. You need to either tell Catalyst (collect statistics) or force the route (broadcast hint).
Technical Answer:
The 50 MB table should trigger a Broadcast Hash Join (default threshold is 10 MB), but it's not happening. Possible reasons:
Statistics are wrong/missing: Spark doesn't know the table is only 50 MB
- Fix:
ANALYZE TABLE small_table COMPUTE STATISTICS
- Or: Set
spark.sql.autoBroadcastJoinThreshold=52428800 (50 MB)
The 50 MB is the size AFTER filters but Spark uses pre-filter size for planning
- Fix: AQE will detect this at runtime and switch to broadcast
Column statistics missing: Without CBO, Spark uses file size
- Fix: Force broadcast:
df1.join(broadcast(df2), "key")
spark.sql("ANALYZE TABLE small_table COMPUTE STATISTICS")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "52428800")
from pyspark.sql.functions import broadcast
result = big_df.join(broadcast(small_df), "key")
spark.conf.set("spark.sql.adaptive.enabled", "true")
Interview Tip: This is one of the most common scenario questions. Walk through all three causes systematically — it shows deep understanding.
What NOT to Say: "Just increase the broadcast threshold to 10 GB." Broadcasting a truly large table will OOM your driver and executors. Only broadcast small tables.
Answer First: 199 workers each have 100 items to pack. But one worker got stuck with 100,000 items — that is data skew. No matter how fast that worker is, they are drowning in work while everyone else is done and idle.
Memory Map: Scenario — Your job has 200 tasks, 199 finish in 2 minutes, but 1 task takes 45 minutes. What's happening -> task-duration outlier signals skew or unhealthy hardware -> input and shuffle metrics separate causes -> repartitioning or executor replacement removes bottleneck -> repeated stage timing verifies balanced completion [01_Spark_Architecture_and_Internals.md:1165].
Q27: Scenario — Your job has 200 tasks, 199 finish in 2 minutes, but 1 task takes 45 minutes. What's happening?
Simple Explanation:
Imagine 200 workers packing boxes. 199 workers each have 100 items to pack. But one worker got stuck with 100,000 items — that is data skew. No matter how fast that worker is, they are drowning in work while everyone else is done and idle.
The solution: either split that worker's pile among multiple workers (AQE skew join, salting), or find a way to reduce the pile before it gets assigned (pre-filtering, broadcast join).
Technical Answer:
This is classic data skew — one partition has disproportionately more data.
How to confirm:
- Spark UI → Stage detail → sort tasks by duration
- Check Input Size/Records for the slow task vs others
- Check Shuffle Read Size for the slow task
Fix options (in order of preference):
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
from pyspark.sql.functions import concat, lit, rand, floor
salt_buckets = 10
skewed_df = skewed_df.withColumn("salted_key",
concat(col("key"), lit("_"), floor(rand() * salt_buckets)))
other_df = other_df.crossJoin(
spark.range(salt_buckets).withColumnRenamed("id", "salt"))
other_df = other_df.withColumn("salted_key",
concat(col("key"), lit("_"), col("salt")))
result = skewed_df.join(other_df, "salted_key")
result = skewed_df.join(broadcast(small_df), "key")
hot_keys = ["key_X", "key_Y"]
skewed_part = df.filter(col("key").isin(hot_keys))
normal_part = df.filter(~col("key").isin(hot_keys))
result = normal_part.join(other_df, "key").union(
skewed_part.join(broadcast(other_df), "key"))
Interview Tip: Data skew is the #1 performance problem in Spark at scale. Be ready to explain salting step-by-step on a whiteboard.
What NOT to Say: "Just add more executors." More executors do not fix skew — the bottleneck is one partition, not total cluster capacity.
Answer First: HDFS is like having filing cabinets in your office — you open a drawer and grab the file instantly. S3 is like a storage locker across town — you have to drive there, wait in line, request your item, and drive back. Every single file access has this overhead.
Memory Map: Scenario — Your Spark job reads from S3 and is 3x slower than reading from HDFS. Why -> remote object storage adds request latency -> connector settings control parallel reads -> file sizing determines request count -> throughput and request metrics validate tuning [01_Spark_Architecture_and_Internals.md:1219].
Q28: Scenario — Your Spark job reads from S3 and is 3x slower than reading from HDFS. Why?
Simple Explanation:
HDFS is like having filing cabinets in your office — you open a drawer and grab the file instantly. S3 is like a storage locker across town — you have to drive there, wait in line, request your item, and drive back. Every single file access has this overhead.
The key differences:
- No data locality: With HDFS, Spark runs tasks where the data lives (reading your own notes = PROCESS_LOCAL). With S3, all reads are remote (calling another office = ANY).
- Slow listing: Finding out what files exist in an S3 "folder" requires multiple API calls. In HDFS, the NameNode knows instantly.
- High per-request latency: Each S3 GET = 50-100 ms. HDFS read = ~1 ms.
Technical Answer:
S3 is an object store, not a filesystem. Key differences:
- List operations are slow: S3 list is O(n) and requires multiple API calls. HDFS metadata is in-memory on NameNode.
- No data locality: With HDFS, tasks run on the node where data resides (PROCESS_LOCAL). With S3, all reads are remote (ANY).
- High latency per request: Each S3 GET has ~50-100 ms latency vs ~1 ms for HDFS
- No append: S3 doesn't support append; each write creates a new object
- Eventual consistency (historically): Though S3 is now strongly consistent for PUTs
Optimizations for S3:
spark.conf.set("spark.hadoop.fs.s3a.connection.maximum", "200")
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")
spark.conf.set("spark.sql.files.openCostInBytes", "0")
- Use Delta Lake (reduces list operations via transaction log)
- Use Auto Loader (file notification mode avoids listing)
Interview Tip: If asked "How do you optimize Spark on S3?", mention: larger partition sizes, more connections, Delta Lake for metadata, and Auto Loader for ingestion.
What NOT to Say: "S3 and HDFS perform the same because they both store files." They have fundamentally different access patterns and latency characteristics.
Answer First: When Spark tries to re-read the recipe (plan the DAG), it has to flip through all 100 pages.
Memory Map: Scenario — You have an iterative ML algorithm that runs 100 iterations. After iteration 50, the job gets extremely slow and eventually fails with StackOverflow. Why -> iterative lineage grows with every derived dataset -> scheduler traversal eventually causes call-stack exhaustion -> checkpoint truncates dependency history -> later iterations reuse bounded lineage [01_Spark_Architecture_and_Internals.md:1255].
Q29: Scenario — You have an iterative ML algorithm that runs 100 iterations. After iteration 50, the job gets extremely slow and eventually fails with StackOverflow. Why?
Simple Explanation:
Think of it like a recipe that says "take the result of step 1, apply step 2, take the result, apply step 3..." After 100 steps, the recipe is 100 pages long. When Spark tries to re-read the recipe (plan the DAG), it has to flip through all 100 pages. Eventually, the recipe book is so thick that Spark's stack overflows trying to hold all the pages open at once.
The fix: every 10 steps, take a photo of the current dish (checkpoint). Now the recipe starts from the photo, not from scratch. The recipe never gets longer than 10 steps.
Technical Answer:
The lineage graph is growing with each iteration. After 100 iterations, the DAG has 100 levels of dependencies. When Spark tries to compute the plan or recover a partition, it traverses this deep lineage → StackOverflowError.
Fix: Checkpoint every N iterations:
spark.sparkContext.setCheckpointDir("/checkpoint/path")
for i in range(100):
rdd = rdd.map(transform_function)
if i % 10 == 0:
rdd.checkpoint()
rdd.count()
Or for DataFrames:
if i % 10 == 0:
df = df.checkpoint()
Interview Tip: This is a classic ML/graph algorithm question. Always mention checkpointing as the solution to growing lineage.
What NOT to Say: "Increase the JVM stack size." That is a band-aid. The lineage will keep growing and eventually overflow any stack size. Checkpointing is the real fix.
Answer First: These two configs control how Spark "slices" files into partitions when reading.
Memory Map: spark.sql.files.maxPartitionBytes and spark.sql.files.openCostInBytes -> maximum file size caps bytes assigned per partition -> open cost estimates per-file scheduling overhead -> Spark packs files into input partitions -> task-size metrics verify balanced reads [01_Spark_Architecture_and_Internals.md:1289].
Q30: What is spark.sql.files.maxPartitionBytes and spark.sql.files.openCostInBytes?
Simple Explanation:
These two configs control how Spark "slices" files into partitions when reading.
maxPartitionBytes = maximum slice size. Like cutting a pizza — each slice cannot be bigger than this (default 128 MB).openCostInBytes = assumed overhead of opening each file. Spark pretends each file is at least this big (default 4 MB) when deciding how to combine files.
The second one is tricky: if you have 10,000 tiny 1 KB files, Spark treats each as 4 MB → 10,000 partitions (too many!). Set it to 0, and Spark combines them aggressively into fewer, larger partitions.
Technical Answer:
maxPartitionBytes (default 128 MB): Maximum size of a partition when reading files. Spark splits large files into partitions of this size.openCostInBytes (default 4 MB): Estimated cost of opening a file. Used to decide when to combine small files into one partition. If you have many tiny files, reducing this to 0 forces more aggressive file combining.
Example: 10,000 files of 1 KB each:
spark.conf.set("spark.sql.files.openCostInBytes", "0")
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")
Interview Tip: This is the go-to answer for "small files problem in Spark." Combine with Delta Lake's OPTIMIZE command for a complete answer.
What NOT to Say: "I repartition after reading to fix the small files problem." That adds a shuffle. Tuning openCostInBytes fixes it at the read stage — no shuffle needed.
SECTION 8: QUICK-FIRE QUESTIONS (Common in Phone Screens)
Answer First: Simple Explanation: Think of transformations as writing a recipe (lazy — nothing happens yet) and actions as turning on the stove (triggers actual cooking). You can write as many recipe steps as you want, but the kitchen stays cold until you hit "cook.".
Memory Map: Transformation vs Action -> transformation vs action clarifies its effect on the lazy execution graph -> transformation extends a lazy graph -> Catalyst sees relational expressions -> action triggers execution -> stages and tasks materialize the result [01_Spark_Architecture_and_Internals.md:1324].
Simple Explanation: Think of transformations as writing a recipe (lazy — nothing happens yet) and actions as turning on the stove (triggers actual cooking). You can write as many recipe steps as you want, but the kitchen stays cold until you hit "cook."
Technical Answer: Transformations are lazy (return new RDD/DF, not computed until action). Actions trigger computation (return value or write to storage). Examples: map/filter = transformation; collect/count/write = action.
Interview Tip: They may follow up with "Why is laziness beneficial?" — answer: it lets Catalyst optimize the full plan globally.
What NOT to Say: "Transformations execute immediately but return a new DataFrame." No — they are lazy and execute ONLY when an action is called.
Answer First: Simple Explanation: Imagine counting votes by state. reduceByKey = each county counts its own votes first, then sends totals to the state office (map-side combine = less data shipped). groupByKey = every county sends ALL individual ballots to the state office, which does all the counting (no pre-aggregation = way more data shipped).
Memory Map: reduceByKey vs groupByKey -> map-side aggregation combines values before exchange -> raw grouping transfers every value -> reduced network volume usually favors the combiner path -> record and byte counts prove the difference [01_Spark_Architecture_and_Internals.md:1334].
Q32: reduceByKey vs groupByKey?
Simple Explanation: Imagine counting votes by state. reduceByKey = each county counts its own votes first, then sends totals to the state office (map-side combine = less data shipped). groupByKey = every county sends ALL individual ballots to the state office, which does all the counting (no pre-aggregation = way more data shipped).
Technical Answer: reduceByKey does a map-side combine (local aggregation before shuffle). groupByKey shuffles ALL data first, then aggregates. reduceByKey always preferred — less data transferred.
Interview Tip: If asked "When would you ever use groupByKey?", the answer is: "Almost never. The only case is when you need ALL values for a key (not an aggregate), and even then, consider combineByKey."
What NOT to Say: "They do the same thing." The shuffle cost difference can be 10x or more.
Answer First: Simple Explanation: RDD = driving with a paper map (you control everything, no optimization). DataFrame = using GPS navigation (Catalyst optimizes the route). Dataset = GPS with voice commands in your native language (typed API, Scala/Java only).
Memory Map: DataFrame vs Dataset vs RDD -> RDD exposes low-level immutable records -> typed Dataset adds encoder-backed JVM types -> DataFrame exposes relational rows to Catalyst -> language and optimizer needs choose the abstraction [01_Spark_Architecture_and_Internals.md:1344].
Q33: DataFrame vs Dataset vs RDD?
Simple Explanation: RDD = driving with a paper map (you control everything, no optimization). DataFrame = using GPS navigation (Catalyst optimizes the route). Dataset = GPS with voice commands in your native language (typed API, Scala/Java only). In PySpark, you only have DataFrame (which is Dataset[Row] under the hood).
Technical Answer: RDD = low-level, no optimization. DataFrame = distributed table, Catalyst-optimized, schema-aware. Dataset = typed DataFrame (Scala/Java only). In PySpark, DataFrame = Dataset[Row]. Always prefer DataFrame over RDD.
Interview Tip: If asked "When would you use RDD?", valid answers: (1) low-level control over partitioning, (2) unstructured data that does not fit a schema, (3) legacy code.
What NOT to Say: "RDDs are faster because they have less overhead." The opposite is true — DataFrames are faster because of Catalyst and Tungsten.
Answer First: Simple Explanation: Imagine a GPS that plans the entire route BEFORE you start driving, vs one that gives you directions one turn at a time. The full-route GPS can find shortcuts and avoid traffic.
Memory Map: lazy evaluation -> transformations record lineage without running tasks -> optimizer sees the complete relational graph -> an action creates executable stages -> unused work can be pruned before scheduling [01_Spark_Architecture_and_Internals.md:1354].
Q34: Why lazy evaluation?
Simple Explanation: Imagine a GPS that plans the entire route BEFORE you start driving, vs one that gives you directions one turn at a time. The full-route GPS can find shortcuts and avoid traffic. That is what lazy evaluation gives Catalyst — a complete view of all operations so it can optimize globally.
Technical Answer: Enables Catalyst to see the full plan before executing → allows global optimizations (predicate pushdown, join reordering, column pruning). Without lazy evaluation, each operation would execute independently.
Interview Tip: Connect this to Catalyst. Laziness is not just about deferring work — it is about enabling optimization.
What NOT to Say: "Lazy evaluation just delays computation to save resources." The primary benefit is optimization, not resource savings.
Answer First: Simple Explanation: A new stage starts whenever cars need to merge and switch lanes (a shuffle). Within a stage, all cars stay in their lane (narrow transformations, pipelined). The shuffle is the toll plaza that separates stages.
Memory Map: What triggers a new stage -> a wide dependency introduces an exchange boundary -> upstream work must finish before downstream fetches -> scheduler cuts the graph at that boundary -> UI stage edges confirm the split [01_Spark_Architecture_and_Internals.md:1364].
Q35: What triggers a new stage?
Simple Explanation: A new stage starts whenever cars need to merge and switch lanes (a shuffle). Within a stage, all cars stay in their lane (narrow transformations, pipelined). The shuffle is the toll plaza that separates stages.
Technical Answer: A wide dependency (shuffle). Each shuffle boundary creates a new stage. All narrow transformations are pipelined within a single stage.
Interview Tip: Follow up with: "How do I know how many stages my job will have?" Answer: count the shuffles (groupBy, join, repartition) + 1.
What NOT to Say: "Each transformation creates a new stage." No — narrow transformations are pipelined within one stage.
Answer First: Simple Explanation: Two different knobs for two different engines. shuffle.partitions controls the DataFrame engine (SQL operations). If you are using DataFrames (you should be), focus on shuffle.partitions.
Memory Map: spark.sql.shuffle.partitions vs spark.default.parallelism -> SQL exchanges use the configured reducer count -> RDD operations derive defaults from available parallelism -> API and operator determine which knob applies -> physical plan confirms the chosen partition count [01_Spark_Architecture_and_Internals.md:1374].
Q36: spark.sql.shuffle.partitions vs spark.default.parallelism?
Simple Explanation: Two different knobs for two different engines. shuffle.partitions controls the DataFrame engine (SQL operations). default.parallelism controls the RDD engine. If you are using DataFrames (you should be), focus on shuffle.partitions.
Technical Answer: shuffle.partitions (default 200) = for DataFrame shuffle operations. default.parallelism = for RDD operations (default = total cores). They are independent settings.
Interview Tip: With AQE, shuffle.partitions is less critical — set it high and let AQE coalesce.
What NOT to Say: "They are the same thing." They control completely different subsystems.
Answer First: Simple Explanation: Think of a restaurant: a Job = one customer order (triggered by one action like count() ). A Stage = one course of the meal (appetizer, main, dessert — separated by shuffles). A Task = one dish within a course (one partition of work).
Memory Map: a task, stage, job -> one action creates a job -> exchange boundaries divide it into stages -> each partition becomes a task -> scheduler events show the resulting hierarchy [01_Spark_Architecture_and_Internals.md:1384].
Q37: What is a task, stage, job?
Simple Explanation: Think of a restaurant: a Job = one customer order (triggered by one action like count()). A Stage = one course of the meal (appetizer, main, dessert — separated by shuffles). A Task = one dish within a course (one partition of work).
Technical Answer:
- Job = one action (e.g.,
count(), write())
- Stage = set of tasks that can run in parallel without shuffle
- Task = one unit of work on one partition
Interview Tip: They may ask "How many tasks will this job have?" Answer: number of partitions in the largest stage.
What NOT to Say: "A task is the same as a stage." A stage contains many tasks (one per partition).
Answer First: Spark waits 3 seconds before downgrading to a worse locality level.
Memory Map: data locality levels -> scheduler compares task preferences with executor hosts -> process-local and node-local placement avoid transfers -> rack or any placement increases distance -> locality wait and fetch metrics quantify compromise [01_Spark_Architecture_and_Internals.md:1397].
Q38: Explain data locality levels.
Simple Explanation: Think of where you store information relative to you:
PROCESS_LOCAL = reading your own notes on your desk (fastest — data is in the same JVM)NODE_LOCAL = grabbing a file from the filing cabinet next to you (same machine, different JVM)NO_PREF = no preference — data has no location preferenceRACK_LOCAL = walking to the filing cabinet in the next room (same rack, different machine)ANY = calling another office across town (any node — slowest)
Spark waits 3 seconds before downgrading to a worse locality level.
Technical Answer: PROCESS_LOCAL (data in same JVM) > NODE_LOCAL (same node, different JVM) > NO_PREF (no preference) > RACK_LOCAL (same rack) > ANY (any node). Spark waits spark.locality.wait (3s) before downgrading.
Interview Tip: Mention that with cloud storage (S3), everything is ANY — data locality only matters with HDFS.
What NOT to Say: "Data locality does not matter anymore." It still matters significantly for HDFS-based clusters and cached data.
Answer First: Simple Explanation: explain() is like asking your GPS to show you the planned route before driving.
Memory Map: explain() help -> parsed and analyzed trees reveal resolution -> optimized tree reveals rule rewrites -> physical tree reveals joins scans and exchanges -> extended output connects estimates to runtime nodes [01_Spark_Architecture_and_Internals.md:1414].
Q39: How does explain() help?
Simple Explanation: explain() is like asking your GPS to show you the planned route before driving. You can see whether it chose the highway (broadcast join) or side streets (sort-merge join), whether it is avoiding toll roads (predicate pushdown), and whether it is taking unnecessary detours.
Technical Answer: Shows logical and physical plans. df.explain(True) shows all 4 phases: Parsed → Analyzed → Optimized → Physical. df.explain("cost") includes CBO statistics. Look for: join strategy, predicate pushdown, partition pruning, codegen nodes.
df.explain(True)
df.explain("cost")
df.explain("formatted")
Interview Tip: "The first thing I do when debugging a slow query is run explain(True)." This sentence alone tells the interviewer you know what you are doing.
What NOT to Say: "I use explain() to see the output of my query." No — explain() shows the execution plan, not the data.
Answer First: Simple Explanation: Imagine you have a massive warehouse (fact table) and a small catalog (dimension table). You want all items from the catalog where category = "Electronics." Without DPP, Spark scans the ENTIRE warehouse.
Memory Map: Dynamic Partition Pruning (DPP) -> dimension-side filtering produces qualifying partition keys -> runtime subquery sends those keys to the fact scan -> irrelevant directories are skipped -> scan files and bytes prove pruning [01_Spark_Architecture_and_Internals.md:1435].
Q40: What is Dynamic Partition Pruning (DPP)?
Simple Explanation: Imagine you have a massive warehouse (fact table) and a small catalog (dimension table). You want all items from the catalog where category = "Electronics." Without DPP, Spark scans the ENTIRE warehouse. With DPP, Spark first checks the catalog to find which shelves have Electronics, then ONLY visits those shelves in the warehouse.
Technical Answer: (Spark 3.0+) When a fact table joins with a filtered dimension table, DPP pushes the dimension filter result into the fact table scan at runtime.
Without DPP: Full scan of fact_sales → join → filter
With DPP: Spark first computes filtered dim_date IDs → uses them to prune fact_sales partitions during scan
SELECT * FROM fact_sales f
JOIN dim_date d ON f.date_id = d.id
WHERE d.year = 2024;
Config: spark.sql.optimizer.dynamicPartitionPruning.enabled=true (default in Spark 3.x)
Interview Tip: DPP is most effective when the fact table is partitioned by the join key. Mention this requirement.
What NOT to Say: "DPP and predicate pushdown are the same thing." Predicate pushdown is compile-time. DPP is runtime — it uses the result of one query to prune another.
Advanced Spark Operations & Coding
💡 Interview Tip
Focus: DataFrame API, Joins, Windows, Streaming, UDFs, Coding Challenges
Approach: Every topic starts with simple explanation → then interview-level depth
MEMORY MAP: PYSPARK OPERATIONS → JAWS-UC
🧠 PYSPARK OPERATIONS → JAWS-UC
PYSPARK OPERATIONSJAWS-UC
JJoins (5 types + 4 physical strategies)
AAggregations (groupBy, window, pivot, cube)
WWindow Functions (row_number, rank, lag/lead)
SStreaming (Structured Streaming + Auto Loader)
UUDFs (Python UDF, Pandas UDF, why UDFs are slow)
CCoding Patterns (dedup, SCD, top-N, gap detection)
SECTION 1: DATAFRAME API DEEP DIVE
Answer First: You have many columns and you want to transform or pick certain ones.
Memory Map: the differences between select(), withColumn(), and selectExpr()? When is each appropriate -> select projects existing expressions together -> withColumn adds or replaces one named expression -> selectExpr parses SQL expression strings -> plan shape and readability choose the API [02_PySpark_Advanced_Operations.md:27].
Q1: What are the differences between select(), withColumn(), and selectExpr()? When is each appropriate?
Simple Explanation:
Think of a spreadsheet. You have many columns and you want to transform or pick certain ones.
select() = "Give me ONLY these columns" — like highlighting specific columns and copying them to a new sheetwithColumn() = "Keep everything, but add/change ONE column" — like inserting a new column into the existing sheetselectExpr() = "Give me these columns, but let me write SQL for them" — like using formulas in Excel
Technical details:
select(): Projects specific columns. Accepts Column objects or strings. Use when you want a subset of columns or need multiple transformations.withColumn(): Adds or replaces a single column. Returns the full DataFrame with the new/modified column.selectExpr(): Like select() but accepts SQL expression strings. Quick for ad-hoc: selectExpr("*", "col1 + col2 as sum_col").
df.select("name", "age", (col("salary") * 1.1).alias("new_salary"))
df.withColumn("new_salary", col("salary") * 1.1)
df.selectExpr("*", "salary * 1.1 as new_salary", "UPPER(name) as name_upper")
Sample data flow:
Original: | name | age | salary |
| Alice | 30 | 50000 |
select(): | name | age | new_salary | ← only selected columns
| Alice | 30 | 55000 |
withColumn(): | name | age | salary | new_salary | ← ALL columns + new one
| Alice | 30 | 50000 | 55000 |
Interview Tip: If asked "when do you use each?", say: "select() for projections and multiple transforms in one shot, withColumn() for adding a single column while keeping everything, and selectExpr() when you want quick SQL expressions without importing functions."
What NOT to Say: "They're all the same." They have very different impacts on the logical plan and performance (see Q2).
Answer First: Instead of saying "Build a house with 3 rooms, 2 bathrooms, and a kitchen" (one instruction), you say "Build room 1.
Memory Map: is chaining multiple withColumn() calls a performance anti-pattern? What's the fix -> repeated withColumn calls deepen projection lineage -> analyzer repeatedly resolves expanding plans -> one select builds a flat projection -> analysis time and plan depth verify improvement [02_PySpark_Advanced_Operations.md:74].
- Simple Explanation:
Imagine you're giving instructions to a builder. Instead of saying "Build a house with 3 rooms, 2 bathrooms, and a kitchen" (one instruction), you say "Build room
- 1Now add room
- 2Now add room
- 3Now add bathroom 1..." — each instruction creates a new blueprint that wraps around the previous one. After 50 instructions, the builder is drowning in 50 nested blueprints.
That's exactly what happens inside Spark's query planner. Each withColumn() creates a new Project node in the logical plan. 50 calls = 50 nested Project nodes that Catalyst must crawl through.
Technical details:
Each withColumn() creates a new Project node in the logical plan. Chaining 50+ calls creates a deeply nested plan that Catalyst must analyze and optimize:
- Extremely slow query planning (minutes)
- Possible StackOverflowError during plan traversal
Visual: What the logical plan looks like
BAD — 4 withColumn() calls create 4 nested Project nodes:
Project [*, d = ...] ← withColumn("d", ...)
Project [*, c = ...] ← withColumn("c", ...)
Project [*, b = ...] ← withColumn("b", ...)
Project [*, a = ...] ← withColumn("a", ...)
Scan table
GOOD — 1 select() call creates 1 Project node:
Project [*, a = ..., b = ..., c = ..., d = ...] ← single select()
Scan table
Bad:
df = df.withColumn("a", expr("..."))
df = df.withColumn("b", expr("..."))
df = df.withColumn("c", expr("..."))
Good:
df = df.select(
"*",
expr("...").alias("a"),
expr("...").alias("b"),
expr("...").alias("c"),
)
Or using functools.reduce:
from functools import reduce
transforms = [("a", expr("...")), ("b", expr("...")), ("c", expr("..."))]
df = reduce(lambda d, t: d.withColumn(t[0], t[1]), transforms, df)
Interview Tip: This is a VERY common question. Show you know the internal reason (nested Project nodes in the logical plan), not just "it's slow." If you can say "I've seen this cause StackOverflowError in production with 100+ columns," that's even better.
What NOT to Say: "withColumn is always bad." It's fine for 1-5 columns. The anti-pattern is chaining 50+ calls.
Answer First: A join combines two tables based on a matching key — like matching a guest list (Table A) with a seating chart (Table B) to figure out who sits where. Different join types answer different questions about what happens when someone is on one list but not the other.
Memory Map: all types of joins in PySpark and their physical implementations -> logical join semantics determine retained rows -> size and ordering determine broadcast hash or merge execution -> exchanges align unmatched partitions -> physical operators and row counts validate behavior [02_PySpark_Advanced_Operations.md:136].
Q3: Explain all types of joins in PySpark and their physical implementations.
Simple Explanation:
A join combines two tables based on a matching key — like matching a guest list (Table A) with a seating chart (Table B) to figure out who sits where. Different join types answer different questions about what happens when someone is on one list but not the other.
Join Types — Visual with Actual Data:
Let's use two small tables:
| emp_id | name | dept_id | | dept_id | dept_name |
|---|
| 1 | Alice | 10 | | 10 | Engineering |
| 2 | Bob | 20 | | 20 | Marketing |
| 3 | Charlie | 30 | | 40 | Finance |
| 4 | Diana | NULL | | | |
Each join type — what comes out:
🧠 Memory Map
| emp_id | name | dept_id | dept_name |
| 1 | Alice | 10 | Engineering | ← both tables have dept 10
| 2 | Bob | 20 | Marketing | ← both tables have dept 20
# Charlie (dept 30) dropped — no match in departments
# Diana (NULL) dropped — NULL never matches
# Finance (dept 40) dropped — no match in employees
| emp_id | name | dept_id | dept_name |
| 1 | Alice | 10 | Engineering |
| 2 | Bob | 20 | Marketing |
| 3 | Charlie | 30 | NULL | ← no dept 30 in right→NULL
| 4 | Diana | NULL | NULL | ← NULL key→no match → NULL
| emp_id | name | dept_id | dept_name |
| 1 | Alice | 10 | Engineering |
| 2 | Bob | 20 | Marketing |
| NULL | NULL | 40 | Finance | ← no emp with dept 40→NULLs
| emp_id | name | dept_id | dept_name |
| 1 | Alice | 10 | Engineering |
| 2 | Bob | 20 | Marketing |
| 3 | Charlie | 30 | NULL | ← left only
| 4 | Diana | NULL | NULL | ← left only (NULL key)
| NULL | NULL | 40 | Finance | ← right only
| emp_id | name | dept_id |
| 1 | Alice | 10 | ← dept 10 exists in departments
| 2 | Bob | 20 | ← dept 20 exists in departments
# Notice: NO columns from right table appear
| emp_id | name | dept_id |
| 3 | Charlie | 30 | ← dept 30 NOT in departments
| 4 | Diana | NULL | ← NULL NOT in departments
| emp_id | name | dept_id | dept_name |
| 1 | Alice | 10 | Engineering |
| 1 | Alice | 10 | Marketing |
| 1 | Alice | 10 | Finance |
| 2 | Bob | 20 | Engineering |
... (4 employees × 3 departments = 12 rows total)
Physical Implementations (How Spark Actually Executes the Join):
| Strategy | When Used | Shuffle? | Notes |
|---|
| Broadcast Hash Join (BHJ) | One side < 10 MB (default) | NO | Fastest. Small side broadcast to all executors. |
| Sort-Merge Join (SMJ) | Both sides large, equi-join | YES | Default for large-large. Both sides sorted by join key. |
| Shuffle Hash Join | One side significantly smaller | YES | Hash table built from smaller side per partition. |
| Broadcast Nested Loop (BNLJ) | Non-equi join, one side small | NO | Broadcast small side, nested loop. |
| Cartesian Product | Cross join or non-equi, both large | YES | Extremely expensive. Avoid if possible. |
Decision Tree — Which join strategy does Spark pick?
One side < 10 MB? → BROADCAST HASH JOIN (fastest)
Both large + equi-join + keys sorted? → SORT MERGE JOIN (default for large)
Both large + equi-join + keys not sorted? → SHUFFLE HASH JOIN
Non-equi join? → BROADCAST NESTED LOOP or CARTESIAN
Analogies to remember:
- Broadcast join = Teacher distributing handouts to every student (small table sent everywhere). No students need to move — the handout comes to them.
- Sort Merge join = Two people merging sorted card decks. Both decks are sorted by number, you walk through both simultaneously matching pairs. Fast, but you need to sort first.
- Shuffle = Moving furniture between apartments (expensive!). Data has to physically move across the network to land on the right executor.
Force a broadcast:
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")
Interview Tip: Draw the decision tree on the whiteboard. Interviewers love seeing you reason about which physical strategy Spark will pick. Mention that you can check the strategy with df.explain() — the plan will say "BroadcastHashJoin" or "SortMergeJoin."
What NOT to Say: "I always use broadcast join because it's faster." Broadcast join can cause driver OOM if the table is too large. You need to understand WHEN each strategy applies.
Answer First: Spark automatically broadcasts a table if it thinks the table is smaller than 10 MB. The idea is: "If this table is tiny, just send a copy to every executor — no need to shuffle the big table." But this can go wrong in several ways.
Memory Map: the default broadcast join threshold? What are the pitfalls of broadcast joins -> estimated small-side bytes are compared with the threshold -> driver materializes and distributes the relation -> every executor builds a local lookup -> memory pressure and stale estimates expose risk [02_PySpark_Advanced_Operations.md:244].
Q4: What is the default broadcast join threshold? What are the pitfalls of broadcast joins?
Simple Explanation:
Spark automatically broadcasts a table if it thinks the table is smaller than 10 MB. The idea is: "If this table is tiny, just send a copy to every executor — no need to shuffle the big table." But this can go wrong in several ways.
Think of it like sending a printed copy of a document to every employee in the building. If the document is 2 pages, great. If it turns out to be 200 pages, you've just overwhelmed the print room (the driver).
Technical details:
Default: spark.sql.autoBroadcastJoinThreshold = 10 MB (10485760 bytes)
Pitfalls:
- Statistics can be wrong — Spark uses file size, not post-filter size. A 1 GB table filtered down to 1 MB? Spark still sees 1 GB and won't broadcast.
- Driver OOM — Driver collects the broadcast table before sending. If the table is bigger than expected, the driver crashes.
- Memory per executor — Total memory =
table_size x num_executors (broadcast replicated everywhere). A 9 MB table with 100 executors = 900 MB of cluster memory used.
- Dynamic size — Table that was 5 MB yesterday might be 500 MB tomorrow.
When NOT to broadcast:
- When the "small" table size is unpredictable
- When the table is actually large after transformations
- When the driver has limited memory
df.explain(True)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100m")
Interview Tip: If asked about broadcast joins, always mention the driver OOM risk. This shows production experience — it's a classic gotcha that only people who've debugged real pipelines know about.
What NOT to Say: "I set the broadcast threshold to 1 GB for everything." This is dangerous and shows you don't understand the driver memory implications.
Answer First: Data skew means one key has MUCH more data than others. Imagine a post office where 90% of all mail goes to one zip code. That one mail carrier is overwhelmed while the others are idle.
Memory Map: you handle skewed data in a join? Explain ALL techniques -> frequency analysis identifies dominant keys -> salting or skew splitting divides heavy groups -> adaptive handling isolates oversized partitions -> task-size distribution verifies balance [02_PySpark_Advanced_Operations.md:283].
Q5: How do you handle skewed data in a join? Explain ALL techniques.
Simple Explanation:
Data skew means one key has MUCH more data than others. Imagine a post office where 90% of all mail goes to one zip code. That one mail carrier is overwhelmed while the others are idle.
In Spark, this means one partition has millions of rows while others have thousands. The one overloaded partition becomes the bottleneck — the entire job waits for it to finish.
Analogy: Shuffle = Moving furniture between apartments. Skew = 99% of the furniture goes to one apartment while 99 other apartments get almost nothing. That one apartment is overloaded and takes forever.
Technique 1: Salting (Most Common)
from pyspark.sql.functions import lit, rand, floor, explode, array, col
salt_buckets = 10
large_df = large_df.withColumn("salt", floor(rand() * salt_buckets).cast("int"))
small_df = small_df.withColumn(
"salt", explode(array([lit(i) for i in range(salt_buckets)]))
)
result = large_df.join(small_df, ["key", "salt"]).drop("salt")
Before salting vs after:
🧠 Memory Map
Partition 0: key="USA"→10 million rows ← BOTTLENECK
Partition 1: key="UK"→1,000 rows
Partition 2: key="JP"→1,000 rows
Partition 0: key="USA", salt=0→1 million rows ← evenly split!
Partition 1: key="USA", salt=1→1 million rows
...
Partition 9: key="USA", salt=9→1 million rows
Partition 10: key="UK", salt=0→1,000 rows
Technique 2: AQE Skew Join (Easiest)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")
Technique 3: Isolate-and-Union
skewed_keys = ["key1", "key2"]
skewed_result = large_df.filter(col("key").isin(skewed_keys)) \
.join(broadcast(small_df), "key")
normal_result = large_df.filter(~col("key").isin(skewed_keys)) \
.join(small_df, "key")
result = skewed_result.union(normal_result)
Technique 4: Two-Phase Aggregation (for groupBy skew)
from pyspark.sql.functions import concat, lit, floor, rand, split, sum as _sum
salted = df.withColumn("salted_key", concat(col("key"), lit("_"), floor(rand() * 100).cast("string")))
partial = salted.groupBy("salted_key").agg(_sum("value").alias("partial_sum"))
result = partial.withColumn("key", split(col("salted_key"), "_")[0]) \
.groupBy("key").agg(_sum("partial_sum").alias("total_sum"))
Interview Tip: Start with "In Databricks, I'd first enable AQE which handles skew automatically. For extreme cases, I'd use salting." This shows you know the modern approach AND the manual technique.
What NOT to Say: "I'd just increase the number of partitions." Repartitioning doesn't fix skew — if one key has 90% of data, it still lands on one partition regardless of how many partitions you have.
SECTION 2: WINDOW FUNCTIONS
Answer First: Window functions let you do calculations WITHIN groups without collapsing the groups. Think of it as: "Within each department, rank employees by salary" — you still see every employee, but each one now has a rank number.
Memory Map: window functions. What's the difference between row_number(), rank(), and dense_rank() -> partition key defines peers -> ordering establishes sequence -> frame selects visible rows -> analytic value returns without collapsing input [02_PySpark_Advanced_Operations.md:378].
Q6: Explain window functions. What's the difference between row_number(), rank(), and dense_rank()?
Simple Explanation:
Window functions let you do calculations WITHIN groups without collapsing the groups. Think of it as: "Within each department, rank employees by salary" — you still see every employee, but each one now has a rank number.
Analogy: Window functions = "Within each group, rank/number the rows." Like a class of students — you want to rank each student within their own class, not across the whole school.
Technical details:
from pyspark.sql import Window
from pyspark.sql.functions import row_number, rank, dense_rank
w = Window.partitionBy("department").orderBy(col("salary").desc())
Step-by-step data walkthrough:
📋 Overview
| name | department | salary |
|---------|-----------|--------|
| Alice | Eng | 100 |
| Bob | Eng | 90 |
| Charlie | Eng | 90 |
| Diana | Eng | 80 |
| Eve | Sales | 95 |
| Frank | Sales | 85 |
STEP 1 — partitionBy("department"):
Group 1 (Eng): Alice(100), Bob(90), Charlie(90), Diana(80)
Group 2 (Sales): Eve(95), Frank(85)
STEP 2 — orderBy(salary.desc()) within each partition:
Group 1 (Eng): Alice(100) → Bob(90) → Charlie(90) → Diana(80)
Group 2 (Sales): Eve(95) → Frank(85)
STEP 3 — Apply the function:
| name | department | salary | row_number | rank | dense_rank |
|---|
| Alice | Eng | 100 | 1 | 1 | 1 |
| Bob | Eng | 90 | 2 | 2 | 2 |
| Charlie | Eng | 90 | 3 | 2 | 2 |
| Diana | Eng | 80 | 4 | 4 | 3 |
| Eve | Sales | 95 | 1 | 1 | 1 |
| Frank | Sales | 85 | 2 | 2 | 2 |
Key differences with ties (Bob and Charlie both have salary 90):
| Function | Result for ties | Gaps? | Description |
|---|
row_number() | 2, 3 | N/A — always unique | Arbitrary tiebreaker — one gets 2, the other gets 3 |
rank() | 2, 2 → skip to 4 | YES, gaps after ties | Like Olympic medals — two silvers, no bronze |
dense_rank() | 2, 2 → next is 3 | NO gaps | Like counting distinct salary levels |
result = df.select(
"*",
row_number().over(w).alias("row_number"),
rank().over(w).alias("rank"),
dense_rank().over(w).alias("dense_rank"),
)
Interview Tip: "Which one should I use for deduplication?" Always row_number() — it guarantees exactly one row per group (no ties). For "top N by category" where you want ties, use dense_rank().
What NOT to Say: "They're basically the same." Choosing the wrong one in a dedup query can give you duplicate results (rank/dense_rank don't guarantee uniqueness with ties).
Answer First: A running total uses an ordered, unbounded-preceding window; a moving average changes the frame, and percent-of-total uses a partition-wide sum. Reusing one partition and ordering specification can avoid redundant exchanges.
Memory Map: PySpark code to compute running total, 7-day moving average, and percentage of total — all in one pass -> shared partition and order define one window -> frame clauses derive cumulative and moving ranges -> aggregate expressions reuse the same sort -> physical plan confirms one exchange and ordering [02_PySpark_Advanced_Operations.md:452].
Q7: Write PySpark code to compute running total, 7-day moving average, and percentage of total — all in one pass.
Simple Explanation:
Imagine a sales dashboard. For each day, you want to see:
- Running total — "How much have we sold from day 1 until today?"
- 7-day moving average — "What's the average daily sales over the last 7 days?"
- Percentage of total — "What % of the entire year's sales happened today?"
Window functions let you compute ALL of these in a single query — no self-joins, no subqueries.
Technical details:
from pyspark.sql import Window
from pyspark.sql.functions import sum as _sum, avg, col
cumulative_w = Window.partitionBy("category").orderBy("date") \
.rowsBetween(Window.unboundedPreceding, Window.currentRow)
moving_w = Window.partitionBy("category").orderBy("date") \
.rowsBetween(-6, Window.currentRow)
total_w = Window.partitionBy("category")
result = df.select(
"*",
_sum("revenue").over(cumulative_w).alias("running_total"),
avg("revenue").over(moving_w).alias("moving_avg_7d"),
(col("revenue") / _sum("revenue").over(total_w) * 100).alias("pct_of_total")
)
Sample data flow:
| category | date | revenue |
|---|
| Books | 2026-01-01 | 100 |
| Books | 2026-01-02 | 150 |
| Books | 2026-01-03 | 200 |
| Books | 2026-01-04 | 50 |
| category | date | revenue |
| Books | 2026-01-01 | 100 |
| Books | 2026-01-02 | 150 |
| Books | 2026-01-03 | 200 |
| Books | 2026-01-04 | 50 |
Interview Tip: Mention that this runs in ONE pass over the data. Without window functions, you'd need multiple self-joins or subqueries — much slower and harder to read.
What NOT to Say: "I'd use a groupBy and then join back." That works but is far less efficient than window functions for this use case.
Answer First: Both define the "window frame" — which rows the function looks at. The difference is HOW they count.
Memory Map: the difference between rowsBetween and rangeBetween -> rowsBetween counts physical row offsets -> rangeBetween compares ordering-value distances -> duplicate order values change range membership -> boundary examples reveal different frames [02_PySpark_Advanced_Operations.md:516].
Q8: What is the difference between rowsBetween and rangeBetween?
Simple Explanation:
Both define the "window frame" — which rows the function looks at. The difference is HOW they count:
rowsBetween = counts by physical position (row 1, row 2, row 3...)rangeBetween = counts by value (all rows where the value is within a certain range)
Analogy: Imagine you're at a concert. rowsBetween(-2, 0) = "me and the 2 people directly in front of me" (physical seats). rangeBetween(-2, 0) = "me and everyone whose ticket price is within $2 of mine" (based on value, not position).
Technical details:
| date | revenue |
|---|
| 2026-01-01 | 100 |
| 2026-01-02 | 150 |
| 2026-01-04 | 200 |
| 2026-01-05 | 50 |
rowsBetween: Physical offset by row count. -6, 0 = current row and 6 rows before.rangeBetween: Logical offset by value. -6, 0 = current value and values up to 6 less.
Critical difference: With rangeBetween, if your data has gaps (e.g., missing dates), the window adjusts logically. With rowsBetween, it always uses the physical row positions.
from pyspark.sql.functions import unix_timestamp
days_7 = 7 * 86400
w = Window.partitionBy("category") \
.orderBy(unix_timestamp("date")) \
.rangeBetween(-days_7, 0)
Interview Tip: If asked "How do you compute a 7-day moving average when dates have gaps?", use rangeBetween with unix_timestamp. This is a common follow-up that trips people up.
What NOT to Say: "I'd use rowsBetween(-6, 0) for a 7-day window." That only works if you have data for every single day with no gaps.
Answer First: Partition purchases by customer and order them by timestamp. Window ranks identify the first and last purchases, while lag or a second-ranked row supplies the first-to-second interval.
Memory Map: Scenario — Find the first and last purchase per customer, plus the time between their first and second purchase -> customer partition orders purchases by timestamp -> row numbers identify first and last events -> lag exposes the second-purchase interval -> sample customers verify retention measures [02_PySpark_Advanced_Operations.md:571].
Q9: Scenario — Find the first and last purchase per customer, plus the time between their first and second purchase.
Simple Explanation:
For each customer, we want: when did they first buy, when did they last buy, and how many days between purchase #1 and purchase #2? This tells us about customer retention — a short gap means they came back quickly.
Technical details:
from pyspark.sql import Window
from pyspark.sql.functions import first, last, lead, datediff, col, row_number
w = Window.partitionBy("customer_id").orderBy("purchase_date")
result = df.withColumn("purchase_rank", row_number().over(w)) \
.withColumn("first_purchase", first("purchase_date").over(
w.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
)) \
.withColumn("last_purchase", last("purchase_date").over(
w.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
)) \
.withColumn("next_purchase", lead("purchase_date", 1).over(w)) \
.filter(col("purchase_rank") == 1) \
.withColumn("days_to_second_purchase",
datediff(col("next_purchase"), col("purchase_date"))
)
Sample data flow:
| customer_id | purchase_date |
|---|
| C1 | 2026-01-01 |
| C1 | 2026-01-15 |
| C1 | 2026-03-01 |
| customer_id | purchase_date |
| C1 | 2026-01-01 |
| C1 | 2026-01-15 |
| C1 | 2026-03-01 |
| customer_id | first_purchase |
| C1 | 2026-01-01 |
Interview Tip: Mention that lead() looks at the NEXT row while lag() looks at the PREVIOUS row. For "time to second purchase," you need lead from the first row, not lag from the second.
What NOT to Say: "I'd do a self-join." Window functions are far more efficient than self-joins for this pattern.
SECTION 3: PARTITIONING & BUCKETING
Answer First: Both change the number of partitions in your DataFrame. Think of partitions as boxes of data.
Memory Map: repartition() vs coalesce(). When do you use each -> repartition vs coalesce use each determines row placement and downstream task balance -> distribution rule assigns rows -> partition count controls task size -> shuffle or pruning changes runtime work -> task-size spread validates balance [02_PySpark_Advanced_Operations.md:629].
Q10: Explain repartition() vs coalesce(). When do you use each?
Simple Explanation:
Both change the number of partitions in your DataFrame. Think of partitions as boxes of data.
repartition() = Dump ALL boxes out, then redistribute evenly into new boxes. Expensive (requires a full shuffle) but guarantees even distribution.coalesce() = Merge adjacent boxes together WITHOUT moving most of the data. Cheap but can create uneven boxes.
Analogy: Moving into a new house.
repartition(4) = Unpack everything, then repack into exactly 4 evenly-filled boxes. Takes time but organized.coalesce(4) = Just combine nearby boxes. Box 1 stays as is, box 2 gets merged into box 3, etc. Fast but some boxes might be overstuffed.
Technical details:
| Aspect | repartition(n) | coalesce(n) |
|---|
| Shuffle | YES (full shuffle) | NO (narrow transformation) |
| Increase partitions? | Yes | No (only decrease) |
| Even distribution? | Yes | No (merges adjacent partitions, can be uneven) |
| By column? | Yes: repartition(n, "col") | No |
| Use case | Need even distribution, increase partitions, join optimization | Reduce partitions before write |
Repartition by column (hash-partitioned):
df = df.repartition(100, "user_id")
df.coalesce(10).write.format("delta").save("/path")
Interview Tip: The most common use of coalesce() is right before a write to reduce the number of small output files. Say: "After a filter that reduces data by 90%, I coalesce to avoid writing thousands of tiny files."
What NOT to Say: "I use repartition before every write." That triggers an unnecessary shuffle. Use coalesce when reducing partitions.
Answer First: Bucketing is like pre-sorting your filing cabinet by category ONCE, so you never have to sort it again.
Memory Map: bucketing? How does it eliminate shuffles -> write-time hashing assigns rows to stable bucket files -> compatible tables share key distribution -> planner can avoid a later exchange -> bucket count and join plan confirm reuse [02_PySpark_Advanced_Operations.md:668].
Q11: What is bucketing? How does it eliminate shuffles?
Simple Explanation:
Bucketing is like pre-sorting your filing cabinet by category ONCE, so you never have to sort it again. When you save a table, you tell Spark: "Organize this data into N buckets by this column." From then on, any join or groupBy on that column is already organized — no shuffle needed.
Technical details:
Bucketing pre-partitions data into a fixed number of buckets by hash of specified columns, and optionally sorts within each bucket.
df.write.bucketBy(256, "user_id").sortBy("user_id").saveAsTable("bucketed_users")
How it eliminates shuffles:
When two bucketed tables with the same bucket count and bucket column are joined, Spark performs a Sort-Merge Join WITHOUT shuffle — data with the same key is already co-located.
WITHOUT bucketing:
Table A (random order) ──shuffle──┐
├── Sort-Merge Join
Table B (random order) ──shuffle──┘
Time: shuffle A + shuffle B + join = SLOW
WITH bucketing (both tables bucketed by user_id, 256 buckets):
Table A bucket 0 ─┐
Table B bucket 0 ─┘── join (no shuffle, already co-located!)
Table A bucket 1 ─┐
Table B bucket 1 ─┘── join (no shuffle!)
... × 256 buckets
Time: just the join = FAST
Caveats:
- Only works with Hive-managed tables (
saveAsTable, not save)
- Bucket count must match between tables
spark.sql.sources.bucketing.enabled must be true- In Databricks, consider Liquid Clustering as a modern alternative
Interview Tip: Mention Liquid Clustering as the Databricks-native replacement for bucketing + Z-ORDER. It's simpler and auto-tunes. Shows you know modern Databricks features.
What NOT to Say: "I always bucket every table." Bucketing adds overhead on write and only helps if you frequently join on the same column.
Answer First: Hash partitioning = Take the key, run it through a math function, and the result tells you which partition. Like assigning students to classes by last-name hash. Fast, but no ordering.
Memory Map: Hash Partitioning vs Range Partitioning -> hash function spreads keys without preserving order -> ordered bounds keep adjacent values together -> equality workloads favor even distribution -> interval scans benefit from bounded placement [02_PySpark_Advanced_Operations.md:714].
Q12: Compare Hash Partitioning vs Range Partitioning.
Simple Explanation:
Two ways to decide which partition a row goes to:
- Hash partitioning = Take the key, run it through a math function, and the result tells you which partition. Like assigning students to classes by last-name hash. Fast, but no ordering.
- Range partitioning = Split data into ranges (A-F, G-L, M-R, S-Z). Ordered, but requires knowing the data distribution first.
Technical details:
| Aspect | Hash Partitioning | Range Partitioning |
|---|
| Algorithm | partition = hash(key) % numPartitions | Partitions by value ranges (requires sampling) |
| Use case | Equi-joins, groupBy | orderBy/sortBy, range queries |
| Skew risk | Yes, if hash distribution poor (e.g., many nulls) | Can be balanced with good sampling |
| Output | Unordered within partitions | Sorted partitions |
df.repartition(100, "user_id")
df.repartitionByRange(100, "date")
Interview Tip: Hash partitioning is what Spark uses internally for shuffles during joins and groupBy. Range partitioning is what Spark uses for global sorting (orderBy).
What NOT to Say: "Hash partitioning always gives even distribution." If most of your keys hash to the same bucket (e.g., lots of NULLs), you get skew.
Answer First: A UDF (User-Defined Function) is custom Python code that you plug into Spark. Spark runs on the JVM (Java Virtual Machine), but your Python code runs in a separate Python process.
Memory Map: UDFs? Why should you avoid them? What are the alternatives -> Python UDF crosses the JVM boundary -> serialization blocks Catalyst visibility -> built-ins or SQL expressions retain native optimization -> operator timing proves the replacement benefit [02_PySpark_Advanced_Operations.md:748].
Q13: What are UDFs? Why should you avoid them? What are the alternatives?
Simple Explanation:
A UDF (User-Defined Function) is custom Python code that you plug into Spark. The problem? Spark runs on the JVM (Java Virtual Machine), but your Python code runs in a separate Python process. Every row of data has to be shipped from the JVM to Python and back — like sending mail between two buildings.
Analogy:
- Native Spark functions = Walking inside one building (JVM). Everything is fast, the optimizer knows every step you take.
- Python UDF = Sending mail between two buildings (JVM to Python and back). You have to package each letter (serialize), walk it to the other building (socket transfer), unpackage it (deserialize), process it, then package the result and walk it back. For EVERY. SINGLE. ROW.
Serialization diagram — why Python UDFs are slow:
📐 Architecture Diagram
ROW-BY-ROW PYTHON UDF
=====================
JVM (Spark Executor) Python Worker Process
┌──────────────────┐ ┌──────────────────┐
│ │ serialize │ │
│ Row 1 data ────┼──────────►──┼──► process row 1 │
│ │ (pickle) │ (your code) │
│ ◄───────────────┼──────────◄──┼──── result 1 │
│ │ deserialize │ │
│ Row 2 data ────┼──────────►──┼──► process row 2 │
│ │ │ │
│ ◄───────────────┼──────────◄──┼──── result 2 │
│ │ │ │
│ ... × millions │ │ ... × millions │
└──────────────────┘ └──────────────────┘
Overhead per row: serialize + socket transfer + deserialize + Python GIL
Total overhead: O(num_rows) × per-row cost = VERY SLOW
PANDAS UDF (VECTORIZED)
=======================
JVM (Spark Executor) Python Worker Process
┌──────────────────┐ ┌──────────────────┐
│ │ Arrow batch │ │
│ 10,000 rows ────┼──────────►──┼──► pd.Series │
│ (one batch) │ (zero-copy)│ (vectorized ops) │
│ │ │ │
│ ◄───────────────┼──────────◄──┼──── results batch│
│ │ │ │
│ next 10K rows ──┼──────────►──┼──► pd.Series │
└──────────────────┘ └──────────────────┘
Overhead per batch: 1 Arrow transfer (near zero-copy)
Total overhead: O(num_batches) × per-batch cost = MUCH FASTER
Technical details:
Why Python UDFs are slow:
- Data serialized from JVM → Python process (via socket) → back to JVM
- Each row individually processed in Python (no vectorization)
- Catalyst cannot optimize through UDFs (no predicate pushdown, no codegen)
- Python GIL limits true parallelism within a worker
Performance hierarchy (fastest to slowest):
- Built-in Spark SQL functions — Catalyst-optimized, codegen, runs in JVM (1x baseline)
- Pandas UDF (vectorized) — Arrow serialization, batch processing with pandas/numpy (3-5x slower)
mapInPandas — Similar to Pandas UDF, for partition-level processing (5-10x slower)- Row-at-a-time Python UDF — Avoid if possible (10-100x slower)
Interview Tip: If asked "How do you optimize a Python UDF?", say: "First, I try to replace it with built-in functions. If not possible, I convert it to a Pandas UDF for vectorized processing with Apache Arrow. As a last resort, I'd use mapInPandas for partition-level processing."
What NOT to Say: "UDFs are fine, they're just Python functions." This shows you don't understand the JVM-Python serialization overhead that makes them 10-100x slower.
Answer First: Pandas UDFs come in 3 flavors, depending on what goes IN and what comes OUT.
Memory Map: a Pandas UDF. When do you use SCALAR vs GROUPED_MAP vs GROUPED_AGG -> Arrow transfers vector batches into Python -> scalar form returns one value per input row -> grouped-map form returns a frame per group -> grouped aggregate returns one value per group [02_PySpark_Advanced_Operations.md:817].
Q14: Write a Pandas UDF. When do you use SCALAR vs GROUPED_MAP vs GROUPED_AGG?
Simple Explanation:
Pandas UDFs come in 3 flavors, depending on what goes IN and what comes OUT:
- Scalar = Column in → Column out (like a regular Spark function but written in pandas)
- Grouped Map = Group of rows in → Group of rows out (for per-group processing like ML)
- Grouped Aggregate = Group of rows in → Single value out (for custom aggregations)
Technical details:
Scalar Pandas UDF (column → column):
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd
@pandas_udf(DoubleType())
def normalize(s: pd.Series) -> pd.Series:
return (s - s.mean()) / s.std()
df = df.withColumn("normalized_salary", normalize(col("salary")))
Grouped Map (group → DataFrame):
from pyspark.sql.functions import pandas_udf, PandasUDFType
@pandas_udf(output_schema, PandasUDFType.GROUPED_MAP)
def train_model(pdf: pd.DataFrame) -> pd.DataFrame:
model = LinearRegression().fit(pdf[features], pdf[target])
pdf["prediction"] = model.predict(pdf[features])
return pdf
result = df.groupBy("category").apply(train_model)
Grouped Aggregate (group → scalar):
@pandas_udf(DoubleType(), PandasUDFType.GROUPED_AGG)
def weighted_mean(values: pd.Series, weights: pd.Series) -> float:
return (values * weights).sum() / weights.sum()
result = df.groupBy("category").agg(weighted_mean(col("value"), col("weight")))
When to use which:
🧠 Memory Map
Need to transform a column?→SCALAR
Example: normalize values, parse strings, custom math
Need to process entire groups?→GROUPED_MAP
Example: train ML model per group, custom time-series per group
Need a custom aggregate?→GROUPED_AGG
Example: weighted mean, trimmed mean, custom statistical functions
Interview Tip: In modern PySpark (3.0+), the decorator syntax is preferred. Mention that Scalar Pandas UDFs are the most common and give the best performance improvement over regular UDFs (3-100x faster due to Arrow serialization).
What NOT to Say: "I'd write a regular Python UDF instead." Always prefer Pandas UDFs when you must use custom Python logic.
Answer First: mapInPandas lets you process entire partitions of data using pandas. Instead of getting one column or one group, you get an iterator of pandas DataFrames — each representing a chunk of the partition. This is perfect for "load a model once, score every row" patterns.
Memory Map: mapInPandas and when do you use it -> each partition becomes an iterator of pandas frames -> Python code may emit zero or many frames -> schema contract validates returned batches -> worker memory bounds safe partition size [02_PySpark_Advanced_Operations.md:886].
Q15: What is mapInPandas and when do you use it?
Simple Explanation:
mapInPandas lets you process entire partitions of data using pandas. Instead of getting one column or one group, you get an iterator of pandas DataFrames — each representing a chunk of the partition. This is perfect for "load a model once, score every row" patterns.
Analogy: Instead of sending each letter individually (row-at-a-time UDF) or even in batches (Pandas UDF), you send the entire mailbag to the Python worker and let it process everything at once. The model is loaded ONCE per partition, not per row or per batch.
Technical details:
def predict_batch(iterator):
import pickle
model = pickle.load(open("/dbfs/models/model.pkl", "rb"))
for batch_df in iterator:
batch_df["prediction"] = model.predict(batch_df[feature_cols])
yield batch_df
result = spark_df.mapInPandas(predict_batch, schema=output_schema)
Use when:
- You need custom Python logic that can't be expressed with built-in functions
- Processing entire partitions (load model once, apply to all rows)
- Need pandas/numpy for complex computations
- You want to avoid the overhead of loading resources per batch
Interview Tip: Mention that mapInPandas uses Apache Arrow for zero-copy transfer between JVM and Python, making it much faster than traditional mapPartitions with manual serialization.
What NOT to Say: "I'd use a regular UDF to load the model and score each row." Loading a model per row is catastrophically slow.
SECTION 5: CACHING & CHECKPOINTING
Answer First: When you reuse a DataFrame multiple times, Spark normally recomputes it from scratch each time (lazy evaluation). Caching, persisting, and checkpointing are ways to say "save this result so we don't have to recompute it.".
Memory Map: cache(), persist(), and checkpoint(). When would you use each -> cache persist and checkpoint would use each sets the durability and recomputation trade-off -> reused computation identifies materialization value -> storage level or checkpoint sets durability -> partitions are materialized -> reuse and eviction metrics justify the choice [02_PySpark_Advanced_Operations.md:923].
Q16: Explain cache(), persist(), and checkpoint(). When would you use each?
Simple Explanation:
When you reuse a DataFrame multiple times, Spark normally recomputes it from scratch each time (lazy evaluation). Caching, persisting, and checkpointing are ways to say "save this result so we don't have to recompute it."
Analogy:
cache() = Bookmarking a page in a book. Quick to find again, but if the book gets damaged (executor fails), you lose the bookmark and must search again.persist() = Same as cache, but you can choose WHERE to save (memory, disk, or both). Like choosing between a sticky note (memory), a physical bookmark (disk), or both.checkpoint() = Photocopying the page and storing the copy in a safe (HDFS/S3). Even if the original book is destroyed, you still have the copy. Also, you don't need to remember which chapter you were in — the copy is standalone.
Technical details:
| Method | Storage | Lineage | Fault Tolerant | Use Case |
|---|
cache() | MEMORY_AND_DISK | Preserved | No (recompute) | Reused DataFrame |
persist(MEMORY_ONLY) | Memory only | Preserved | No | Fits in memory, reused often |
persist(DISK_ONLY) | Disk only | Preserved | No | Large data, infrequent reuse |
persist(MEMORY_AND_DISK_SER) | Memory (serialized) + disk | Preserved | No | Memory-constrained |
checkpoint() | Reliable storage (HDFS/S3) | Truncated | Yes | Long lineage, iterative algorithms |
localCheckpoint() | Executor local storage | Truncated | No | Fast lineage break, less reliable |
When to checkpoint vs cache:
- Use
cache() when the DataFrame is reused 2+ times and you want to avoid recomputation
- Use
checkpoint() when the lineage is very deep (iterative algorithms) to prevent StackOverflow
- Always call an action after checkpoint to materialize:
df.checkpoint(); df.count()
filtered_df = raw_df.filter(col("status") == "active") \
.select("user_id", "email", "created_at")
filtered_df.cache()
filtered_df.count()
filtered_df.groupBy("created_at").count().show()
filtered_df.join(other_df, "user_id").show()
filtered_df.unpersist()
Interview Tip: Always mention unpersist(). Forgetting to release cached DataFrames is a common memory leak in production jobs. Also mention that cache() is lazy — the data isn't actually cached until an action triggers it.
What NOT to Say: "I cache everything to make it faster." Over-caching wastes memory and can cause executors to spill to disk, making things SLOWER.
SECTION 6: STRUCTURED STREAMING
Answer First: Structured Streaming models an input stream as an incrementally updated table and executes batch-style transformations on each new range of data. Checkpoints preserve source progress and state across restarts.
Memory Map: the Structured Streaming execution model -> structured streaming execution model controls one checkpointed streaming state transition -> offset and event time enter the micro-batch -> stateful logic applies the watermark boundary -> checkpoint commits progress -> replay and sink checks prove correctness [02_PySpark_Advanced_Operations.md:972].
Q17: Explain the Structured Streaming execution model.
Simple Explanation:
Structured Streaming treats real-time data as a table that keeps growing. Every few seconds (or whatever interval you set), Spark looks at "what new rows arrived?" and processes just those new rows using the exact same code you'd write for a batch query.
Analogy: Imagine a restaurant where orders come in continuously. Instead of waiting until the restaurant closes to count all orders, the manager checks the order list every 10 seconds and processes only the NEW orders since the last check. Same counting method, just applied incrementally.
Technical details:
- The stream is treated as an unbounded table
- Each trigger processes new rows appended to this table
- Uses the same Catalyst optimizer as batch queries
Trigger modes:
| Mode | Behavior |
|---|
trigger(processingTime="10 seconds") | Micro-batch every 10 seconds |
trigger(once=True) | Process all available, stop (deprecated) |
trigger(availableNow=True) | Process all available in multiple micro-batches, stop |
| Continuous (experimental) | Row-by-row, ~1 ms latency, at-least-once only |
result = spark.readStream \
.format("delta") \
.table("bronze_events") \
.filter(col("event_type") == "purchase") \
.groupBy("product_id").count()
result.writeStream \
.format("delta") \
.outputMode("complete") \
.trigger(processingTime="10 seconds") \
.option("checkpointLocation", "/checkpoints/purchase_counts") \
.toTable("silver_purchase_counts")
Interview Tip: Emphasize that the code is nearly identical to batch — "I can prototype in batch, then switch to streaming by changing read to readStream and write to writeStream." This is Structured Streaming's key design principle.
What NOT to Say: "Structured Streaming processes one row at a time." It uses micro-batches (unless using the experimental continuous mode).
Answer First: Append mode emits only finalized new rows, update mode emits rows changed in the batch, and complete mode rewrites the entire result table. Query shape and sink capability determine which modes are valid.
Memory Map: output modes? When is each used -> result mutability defines the sink contract -> immutable completions permit incremental emission -> changing aggregates require replacement records -> full-state sinks accept snapshot rewrites [02_PySpark_Advanced_Operations.md:1016].
Q18: What are output modes? When is each used?
Simple Explanation:
After each micro-batch, Spark needs to know: "What results should I write to the sink?" The three output modes answer this differently:
- Append = "Only write NEW rows that won't change" (like adding new entries to a log)
- Complete = "Rewrite the ENTIRE result table every time" (like refreshing a dashboard)
- Update = "Only write rows that CHANGED" (like updating a leaderboard)
Technical details:
| Mode | Behavior | Works With |
|---|
| Append (default) | Only new rows output | Non-aggregation queries, or aggregations with watermark |
| Complete | Entire result table output | Only with aggregations |
| Update | Only changed rows output | Aggregations (rows whose aggregate value changed) |
🧠 Memory Map
Example: counting events per category
After batch 1: {Electronics: 10, Books: 5}
After batch 2: {Electronics: 15, Books: 5, Clothing: 3}
Append mode: Would write→{Clothing: 3} (only truly new categories)
ERROR! Can't guarantee Electronics won't change.
Complete mode: Would write→{Electronics: 15, Books: 5, Clothing: 3}
Entire result table every time. Safe but expensive.
Update mode: Would write→{Electronics: 15, Clothing: 3}
Only the rows that changed. Books stayed at 5 so it's skipped.
Common mistake: Using append mode with aggregations without watermark → throws error because Spark can't guarantee old rows won't change.
Interview Tip: For most streaming-to-Delta pipelines without aggregation, use append. For aggregation dashboards, use update (more efficient than complete). Only use complete when the sink needs the full picture every time.
What NOT to Say: "I always use complete mode." Complete rewrites the entire result every micro-batch — extremely inefficient for large state.
Answer First: A click that happened at 3:00 PM might only arrive at the server at 3:25 PM (due to network delays, offline devices, etc.). Watermarking tells Spark: "Wait up to X minutes for late data.
Memory Map: watermarking with a real scenario -> event timestamps advance the maximum observed time -> allowed delay derives an eviction boundary -> older state can be removed -> records beyond that boundary follow late-data semantics [02_PySpark_Advanced_Operations.md:1057].
Q19: Explain watermarking with a real scenario.
Simple Explanation:
In the real world, data arrives late. A click that happened at 3:00 PM might only arrive at the server at 3:25 PM (due to network delays, offline devices, etc.). Watermarking tells Spark: "Wait up to X minutes for late data. After that, stop waiting and clean up."
Analogy: A professor has a homework deadline of Monday 5 PM. But they accept late submissions up to 30 minutes (the watermark). At 5:30 PM, they stop accepting papers and grade what they have. Without this cutoff, they'd wait forever and never grade anything.
Technical details:
Scenario: Clickstream sessionization. Events may arrive up to 30 minutes late.
from pyspark.sql.functions import window
clicks = spark.readStream.format("kafka").load().select(
col("user_id"),
col("event_time").cast("timestamp"),
col("page_url")
)
sessionized = clicks \
.withWatermark("event_time", "30 minutes") \
.groupBy(
col("user_id"),
window("event_time", "1 hour")
).count()
What watermark does — step by step:
🧠 Memory Map
Time progresses:
3:00 PM — events arrive with event_time = 3:00 PM
max(event_time) = 3:00 PM
watermark = 3:00 - 30 min = 2:30 PM
→ Accept any event with event_time >= 2:30 PM
3:15 PM — events arrive with event_time = 3:15 PM
max(event_time) = 3:15 PM
watermark = 3:15 - 30 min = 2:45 PM
→ Accept any event with event_time >= 2:45 PM
→ Late event with event_time = 2:40 PM? DROPPED! (before watermark)
3:30 PM — max(event_time) = 3:30 PM
watermark = 3:00 PM
→ State for windows ending before 3:00 PM is CLEANED UP
→ This prevents unbounded state growth!
What watermark does (summary):
- Tracks
max(event_time) seen so far
watermark = max(event_time) - 30 minutes- Events with
event_time < watermark are dropped
- State older than watermark is cleaned up (prevents unbounded state growth)
Interview Tip: Always connect watermarks to STATE CLEANUP. The interviewer wants to hear: "Without watermarks, the state store grows forever. Watermarks let Spark know when it's safe to discard old state."
What NOT to Say: "Watermarks guarantee no data is ever lost." Late data arriving AFTER the watermark IS dropped. It's a trade-off between completeness and resource usage.
Answer First: Joining two streams is like matching real-time orders with real-time payments. Both arrive continuously, and you need to hold onto unmatched records from both sides until a match arrives (or until you give up waiting).
Memory Map: stream-stream joins work? What are the requirements -> both inputs require event-time columns -> watermarks bound retained state on each side -> time-range condition limits possible matches -> state metrics reveal whether bounds are effective [02_PySpark_Advanced_Operations.md:1118].
Q20: How do stream-stream joins work? What are the requirements?
Simple Explanation:
Joining two streams is like matching real-time orders with real-time payments. Both arrive continuously, and you need to hold onto unmatched records from both sides until a match arrives (or until you give up waiting).
Technical details:
orders = orders_stream.withWatermark("order_time", "2 hours")
payments = payments_stream.withWatermark("payment_time", "3 hours")
joined = orders.join(
payments,
expr("""
orders.order_id = payments.order_id AND
payments.payment_time BETWEEN orders.order_time AND orders.order_time + interval 1 hour
"""),
"left_outer"
)
Requirements:
- Both sides must have watermarks defined
- Time-range conditions recommended to limit state
- For outer joins: a row is output with nulls once the watermark guarantees no future match is possible
- For inner joins: late data on either side is buffered until watermark allows cleanup
🧠 Memory Map
Without time-range condition:
Spark must buffer ALL unmatched orders and ALL unmatched payments FOREVER
→ State grows unbounded→OOM
With time-range condition (payment within 1 hour of order):
Spark knows: if order_time = 3:00 PM and it's now 4:00 PM,
no payment can possibly match→safe to discard that order from state
→ State is bounded→stable memory usage
Interview Tip: Stream-stream joins are a favorite advanced topic. Mention the state implications: without time bounds and watermarks, state grows forever and the job eventually OOMs.
What NOT to Say: "Stream-stream joins work just like batch joins." They require watermarks and time-range conditions that batch joins don't need.
Answer First: foreachBatch is a bridge between streaming and batch. It says: "For each micro-batch, give me the data as a regular DataFrame — then I'll decide what to do with it." This unlocks batch-only operations (like MERGE) inside a streaming pipeline.
Memory Map: the foreachBatch pattern. When is it needed -> engine hands each micro-batch DataFrame and ID to user code -> batch APIs perform unsupported sink or merge logic -> transaction key makes replay idempotent -> checkpoint restart proves safe re-execution [02_PySpark_Advanced_Operations.md:1167].
Q21: Explain the foreachBatch pattern. When is it needed?
Simple Explanation:
foreachBatch is a bridge between streaming and batch. It says: "For each micro-batch, give me the data as a regular DataFrame — then I'll decide what to do with it." This unlocks batch-only operations (like MERGE) inside a streaming pipeline.
Analogy: Imagine a conveyor belt (the stream) dropping boxes onto a table every 30 seconds. foreachBatch lets you pick up each batch of boxes and do whatever you want with them — sort them, compare with existing inventory, ship some back — things you can't do while they're on the moving belt.
Technical details:
def upsert_to_delta(batch_df, batch_id):
target = DeltaTable.forName(spark, "silver_orders")
target.alias("t").merge(
batch_df.alias("s"),
"t.order_id = s.order_id"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
spark.readStream.table("bronze_orders") \
.writeStream \
.foreachBatch(upsert_to_delta) \
.option("checkpointLocation", "/checkpoints/silver_orders") \
.trigger(processingTime="1 minute") \
.start()
Use when:
- MERGE into Delta Lake (can't do with regular streaming write)
- Writing to multiple sinks in one pipeline
- Calling external APIs per batch
- Complex deduplication logic
- Any operation that needs the full batch as a DataFrame
Interview Tip: foreachBatch is the answer to "How do you do streaming MERGE/upsert in Databricks?" — the most common streaming interview question. Always mention using batch_id for idempotency.
What NOT to Say: "I'd stop the stream, run a batch MERGE, then restart." That defeats the purpose of streaming.
Answer First: "Exactly-once" means every record is processed exactly one time — not zero, not twice. This is surprisingly hard in distributed systems. Spark achieves it by combining three things: a replayable source, checkpointing, and an idempotent sink.
Memory Map: you achieve exactly-once semantics in Structured Streaming -> source offsets persist in checkpoint state -> deterministic processing recreates a batch -> idempotent or transactional sink rejects duplicate commit IDs -> restart reconciliation proves one visible result [02_PySpark_Advanced_Operations.md:1210].
Q22: How do you achieve exactly-once semantics in Structured Streaming?
Simple Explanation:
"Exactly-once" means every record is processed exactly one time — not zero, not twice. This is surprisingly hard in distributed systems. Spark achieves it by combining three things: a replayable source, checkpointing, and an idempotent sink.
Analogy: Imagine processing bank transactions:
- Source must be replayable — like a numbered list where you can say "start from transaction #500" (Kafka offsets)
- Engine must remember where it stopped — "I finished up to transaction #499" (checkpointing)
- Sink must handle duplicates — "If I accidentally process #499 again, the result is the same" (idempotent writes)
Technical details:
Three requirements:
- Source: Must be replayable (Kafka with offsets, file source with checkpoints)
- Engine: Checkpointing tracks offsets and state. On restart, Spark replays from last committed offset.
- Sink: Must be idempotent (re-writing the same batch produces the same result)
Built-in exactly-once sinks:
- Delta Lake (ACID transactions)
- File sink (uses batch ID in file names)
- Kafka sink (with idempotent producer)
def idempotent_write(batch_df, batch_id):
batch_df.write.format("delta") \
.mode("overwrite") \
.option("replaceWhere", f"batch_id = {batch_id}") \
.save("/path/to/output")
Interview Tip: The key insight is that exactly-once is achieved END-TO-END, not by any single component. Source + Engine + Sink must all cooperate. If any one fails (e.g., a non-idempotent sink), you lose the guarantee.
What NOT to Say: "Spark guarantees exactly-once automatically." Only true with the right source + sink combination. A write to a REST API without idempotency keys is NOT exactly-once.
Answer First: The state store holds intermediate data that the streaming query needs to remember (e.g., running counts, unmatched join records, dedup history). If it grows forever, your job will eventually run out of memory and crash.
Memory Map: Scenario — Your streaming pipeline's state store is growing unbounded. How do you fix it -> stateful keys accumulate without an eviction condition -> event-time bounds make old entries removable -> aggregation or join logic applies the boundary -> state-row metrics confirm bounded growth [02_PySpark_Advanced_Operations.md:1249].
Q23: Scenario — Your streaming pipeline's state store is growing unbounded. How do you fix it?
Simple Explanation:
The state store holds intermediate data that the streaming query needs to remember (e.g., running counts, unmatched join records, dedup history). If it grows forever, your job will eventually run out of memory and crash.
This is like a to-do list that only adds items but never removes completed ones — eventually the list is so long you can't carry it.
Technical details:
- Add watermarks to bound the state
- Add time constraints on joins to limit buffered data
- Use RocksDB state store (disk-based, handles large state):
spark.conf.set(
"spark.sql.streaming.stateStore.providerClass",
"org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
)
- Set
spark.sql.streaming.stateStore.minDeltasForSnapshot for compaction
- Monitor state via StreamingQueryListener:
class StateMonitor(StreamingQueryListener):
def onQueryProgress(self, event):
state_info = event.progress.stateOperators
for op in state_info:
print(f"State rows: {op.numRowsTotal}, Memory: {op.memoryUsedBytes}")
Checklist for debugging unbounded state:
Are watermarks defined?
NO → Add withWatermark() on the event-time column
YES → Is the watermark delay too large? Reduce it.
Are join conditions time-bounded?
NO → Add "BETWEEN order_time AND order_time + interval X"
YES → Is the interval too large?
Are you using dropDuplicates without watermark?
YES → Add watermark so Spark can expire old dedup state
Is the state store in-memory?
YES → Switch to RocksDB state store for disk-based state
Interview Tip: Start with watermarks (the most common fix), then mention RocksDB (shows deep knowledge). If the interviewer asks about monitoring, mention StreamingQueryListener.
What NOT to Say: "I'd just increase executor memory." That's a band-aid, not a fix. The state will keep growing and eventually exceed any amount of memory.
Answer First: Both are for "batch-style streaming" — process all available data, then stop. The difference is HOW they process.
Memory Map: the difference between trigger(once=True) and trigger(availableNow=True) -> once trigger processes one micro-batch then stops -> available-now trigger advances through all pending batches -> checkpoint preserves progress between batches -> backlog tests expose semantic difference [02_PySpark_Advanced_Operations.md:1301].
Q24: What is the difference between trigger(once=True) and trigger(availableNow=True)?
Simple Explanation:
Both are for "batch-style streaming" — process all available data, then stop. The difference is HOW they process:
trigger(once=True) = Shoves ALL data into ONE micro-batch. Can OOM if there's a lot of data.trigger(availableNow=True) = Splits data into MULTIPLE micro-batches, processes them sequentially, then stops. Memory-friendly.
Analogy: You have 1000 emails to process. once=True = try to open all 1000 at once (your computer might crash). availableNow=True = open 100 at a time, 10 batches, then stop (much safer).
Technical details:
| Aspect | trigger(once=True) | trigger(availableNow=True) |
|---|
| Processing | One single micro-batch | Multiple micro-batches |
| Parallelism | All data in one batch | Spreads across multiple batches |
| Memory | Can OOM on large backlog | More memory-friendly |
| Status | Deprecated (Spark 3.3+) | Recommended replacement |
| Use case | Periodic batch-style runs | Periodic batch-style runs (better) |
df.writeStream.trigger(once=True).start()
df.writeStream.trigger(availableNow=True).start()
Interview Tip: Always recommend availableNow=True over once=True. It shows you know the modern API and understand the memory implications. Common pattern: schedule this via Databricks Jobs to run every hour for cost-efficient near-real-time processing.
What NOT to Say: "They're the same thing." The memory behavior difference is significant for production workloads.
SECTION 7: CODING CHALLENGES
Answer First: You have duplicate records for the same ID (e.g., multiple updates to the same user). You want to keep only the LATEST version of each record. This is the most common PySpark interview coding question.
Memory Map: Deduplicate records keeping the most recent per key -> business key partitions duplicate candidates -> descending event order ranks newest rows -> row-number filter retains one record -> uniqueness and freshness assertions confirm output [02_PySpark_Advanced_Operations.md:1341].
Q25: Deduplicate records keeping the most recent per key.
Simple Explanation:
You have duplicate records for the same ID (e.g., multiple updates to the same user). You want to keep only the LATEST version of each record. This is the most common PySpark interview coding question.
Technical details:
from pyspark.sql.functions import row_number, col
from pyspark.sql import Window
window = Window.partitionBy("id").orderBy(col("updated_at").desc())
deduped = df.withColumn("rn", row_number().over(window)) \
.filter(col("rn") == 1) \
.drop("rn")
Sample data flow:
| id | name | updated_at |
|---|
| 1 | Alice | 2026-01-01 |
| 1 | Alice_v2 | 2026-01-15 |
| 2 | Bob | 2026-01-10 |
| id | name | updated_at |
| 1 | Alice_v2 | 2026-01-15 |
| 1 | Alice | 2026-01-01 |
| 2 | Bob | 2026-01-10 |
| id | name | updated_at |
| 1 | Alice_v2 | 2026-01-15 |
| 2 | Bob | 2026-01-10 |
Interview Tip: Use row_number(), NOT rank() or dense_rank(). Those can give ties, meaning you'd keep multiple rows per key. row_number() guarantees exactly one row per partition.
What NOT to Say: "I'd use dropDuplicates(['id'])." That keeps an ARBITRARY record, not necessarily the most recent one. You can't control which record is kept.
Answer First: Pivoting turns unique values in a column into separate columns. Like turning a vertical list of sales-by-category into a horizontal spreadsheet with one column per category.
Memory Map: Pivot a table — convert rows to columns -> group keys establish output rows -> pivot values become named columns -> aggregate resolves collisions within each cell -> expected schema and totals validate reshaping [02_PySpark_Advanced_Operations.md:1391].
Q26: Pivot a table — convert rows to columns.
Simple Explanation:
Pivoting turns unique values in a column into separate columns. Like turning a vertical list of sales-by-category into a horizontal spreadsheet with one column per category.
Technical details:
pivoted = df.groupBy("date") \
.pivot("category", ["Electronics", "Clothing", "Food"]) \
.agg(sum("amount"))
Sample data flow:
| date | category | amount |
|---|
| 2026-01-01 | Electronics | 100 |
| 2026-01-01 | Clothing | 50 |
| 2026-01-01 | Food | 30 |
| 2026-01-02 | Electronics | 120 |
| 2026-01-02 | Food | 40 |
| date | Electronics | Clothing |
| 2026-01-01 | 100 | 50 |
| 2026-01-02 | 120 | NULL |
Follow-up: How do you unpivot (melt)?
from pyspark.sql.functions import expr
unpivoted = df.unpivot("date", ["Electronics", "Clothing", "Food"], "category", "amount")
unpivoted = df.selectExpr(
"date",
"stack(3, 'Electronics', Electronics, 'Clothing', Clothing, 'Food', Food) as (category, amount)"
)
Interview Tip: Always pass the explicit list of values to pivot(). Without it, Spark scans the entire dataset first to discover distinct values — an expensive extra job.
What NOT to Say: "I'd use a for-loop to create separate DataFrames and union them." That's the anti-pattern that pivot was designed to replace.
Answer First: Given a series of IDs (1, 2, 3, 5, 6, 10), find where the gaps are (4 is missing, 7-8-9 are missing). This is useful for audit logs, sequence validation, and data quality checks.
Memory Map: Find gaps in a sequential series -> ordered sequence pairs each value with predecessor -> lag exposes expected next value -> non-unit differences mark missing intervals -> boundary expansion lists actual gaps [02_PySpark_Advanced_Operations.md:1446].
Q27: Find gaps in a sequential series.
Simple Explanation:
Given a series of IDs (1, 2, 3, 5, 6, 10), find where the gaps are (4 is missing, 7-8-9 are missing). This is useful for audit logs, sequence validation, and data quality checks.
Technical details:
from pyspark.sql.functions import lead
from pyspark.sql import Window
w = Window.orderBy("sequence_id")
gaps = df.withColumn("next_id", lead("sequence_id").over(w)) \
.filter(col("next_id") - col("sequence_id") > 1) \
.select(
col("sequence_id").alias("gap_start"),
col("next_id").alias("gap_end"),
(col("next_id") - col("sequence_id") - 1).alias("gap_size")
)
Sample data flow:
| 5 | ← gap: 4 is missing |
|---|
| 10 | ← gap: 7,8,9 are missing |
| sequence_id | next_id |
| 1 | 2 |
| 2 | 3 |
| 3 | 5 |
| 5 | 6 |
| 6 | 10 |
| 10 | NULL |
| gap_start | gap_end |
| 3 | 5 |
| 6 | 10 |
Interview Tip: This pattern (lead/lag + filter) is the standard approach for detecting gaps, islands, and consecutive sequences. Master it — it appears in many variations.
What NOT to Say: "I'd generate all numbers and do a left anti join." That works but is wasteful for sparse sequences with large ranges.
Answer First: Sessionization groups user clicks into "sessions." A new session starts when the user has been inactive for more than X minutes (e.g., 30 minutes). If Alice clicks at 1:00, 1:05, 1:10, then 3:00 — the first three clicks are session 1, and 3:00 starts session 2.
Memory Map: Sessionize clickstream data (gap-based sessions) -> user events sort by occurrence time -> lag measures inactivity between clicks -> threshold crossings start new sessions -> cumulative sum assigns stable session identifiers [02_PySpark_Advanced_Operations.md:1504].
Q28: Sessionize clickstream data (gap-based sessions).
Simple Explanation:
Sessionization groups user clicks into "sessions." A new session starts when the user has been inactive for more than X minutes (e.g., 30 minutes). If Alice clicks at 1:00, 1:05, 1:10, then 3:00 — the first three clicks are session 1, and 3:00 starts session 2.
Technical details:
from pyspark.sql.functions import lag, when, unix_timestamp, sum as _sum, monotonically_increasing_id
from pyspark.sql import Window
w = Window.partitionBy("user_id").orderBy("event_time")
session_timeout = 30 * 60
df = df.withColumn("prev_time", lag("event_time").over(w)) \
.withColumn("new_session",
when(
(unix_timestamp("event_time") - unix_timestamp("prev_time")) > session_timeout, 1
).when(col("prev_time").isNull(), 1)
.otherwise(0)
) \
.withColumn("session_id",
_sum("new_session").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow))
)
Sample data flow:
📋 Overview
| user_id | event_time | page |
|---------|-----------|-----------|
| Alice | 1:00 PM | /home |
| Alice | 1:05 PM | /products |
| Alice | 1:10 PM | /cart |
| Alice | 3:00 PM | /home | ← 110 min gap > 30 min timeout
| Alice | 3:02 PM | /checkout |
STEP 1 — lag() gets previous event time:
| event_time | prev_time | gap_minutes |
|-----------|-----------|-------------|
| 1:00 PM | NULL | N/A |
| 1:05 PM | 1:00 PM | 5 |
| 1:10 PM | 1:05 PM | 5 |
| 3:00 PM | 1:10 PM | 110 | ← > 30 min!
| 3:02 PM | 3:00 PM | 2 |
STEP 2 — flag new sessions:
| event_time | new_session |
|-----------|-------------|
| 1:00 PM | 1 | ← first event = new session
| 1:05 PM | 0 | ← 5 min gap < 30 min
| 1:10 PM | 0 |
| 3:00 PM | 1 | ← 110 min gap > 30 min = NEW SESSION
| 3:02 PM | 0 |
STEP 3 — running sum gives session IDs:
| event_time | new_session | session_id |
|-----------|-------------|------------|
| 1:00 PM | 1 | 1 |
| 1:05 PM | 0 | 1 |
| 1:10 PM | 0 | 1 |
| 3:00 PM | 1 | 2 | ← new session!
| 3:02 PM | 0 | 2 |
Interview Tip: The key insight is using a running sum over boolean flags. This "cumulative sum of flags" pattern appears in many problems beyond sessionization (islands-and-gaps, state changes, etc.).
What NOT to Say: "I'd use a UDF to loop through rows and assign session IDs." Window functions do this natively and much faster.
Answer First: JSON data often has nested objects and arrays. "Flattening" means turning the nested structure into a simple table with one row per leaf-level record. Each explode() turns an array into multiple rows.
Memory Map: Flatten a deeply nested JSON structure -> schema inspection locates struct and array nesting -> struct expansion projects child fields -> explode creates rows for array elements -> row counts and null cases validate flattening [02_PySpark_Advanced_Operations.md:1576].
Q29: Flatten a deeply nested JSON structure.
Simple Explanation:
JSON data often has nested objects and arrays. "Flattening" means turning the nested structure into a simple table with one row per leaf-level record. Each explode() turns an array into multiple rows.
Technical details:
from pyspark.sql.functions import explode, col
raw = spark.read.json("/path/to/nested.json")
flat = raw.select("id", explode("orders").alias("order")) \
.select(
"id",
col("order.order_id"),
col("order.amount"),
explode("order.items").alias("item")
).select(
"id",
"order_id",
"amount",
col("item.product").alias("product_name"),
col("item.qty").alias("quantity")
)
Sample data flow:
| id | order.order_id | order.amount | order.items |
|---|
| C1 | O1 | 100 | [{Laptop,1},{Mouse,2}] |
| C1 | O2 | 50 | [{Book,3}] |
| id | order_id | amount | product_name |
| C1 | O1 | 100 | Laptop |
| C1 | O1 | 100 | Mouse |
| C1 | O2 | 50 | Book |
Generic recursive flattener:
from pyspark.sql.types import StructType, ArrayType
def flatten_df(df):
"""Recursively flatten all nested structs and arrays."""
flat_cols = []
for field in df.schema.fields:
if isinstance(field.dataType, StructType):
for subfield in field.dataType.fields:
flat_cols.append(col(f"{field.name}.{subfield.name}").alias(f"{field.name}_{subfield.name}"))
elif isinstance(field.dataType, ArrayType):
df = df.withColumn(field.name, explode(col(field.name)))
return flatten_df(df)
else:
flat_cols.append(col(field.name))
return df.select(flat_cols)
Interview Tip: Mention that explode creates a new row per array element, which multiplies the row count. For large arrays, this can cause data explosion. Consider using posexplode if you need the array index too.
What NOT to Say: "I'd convert to pandas and flatten there." That defeats the purpose of distributed processing and will fail on large datasets.
Answer First: Classic "top-N per group" pattern. First aggregate revenue per product, then rank within each category, then filter to keep only the top 3.
Memory Map: a query to find the top 3 products by revenue in each category -> line-item amounts aggregate by item and group -> window ordering sorts totals within each group -> dense rank preserves tied leaders -> position filter retains the first three ranks [02_PySpark_Advanced_Operations.md:1659].
Q30: Write a query to find the top 3 products by revenue in each category.
Simple Explanation:
Classic "top-N per group" pattern. First aggregate revenue per product, then rank within each category, then filter to keep only the top 3.
Technical details:
from pyspark.sql.functions import dense_rank, col, sum as _sum
from pyspark.sql import Window
product_revenue = df.groupBy("category", "product_id") \
.agg(_sum("revenue").alias("total_revenue"))
w = Window.partitionBy("category").orderBy(col("total_revenue").desc())
top3 = product_revenue.withColumn("rnk", dense_rank().over(w)) \
.filter(col("rnk") <= 3) \
.drop("rnk")
Sample data flow:
| category | product_id | total_revenue |
|---|
| Electronics | P1 | 5000 |
| Electronics | P2 | 3000 |
| Electronics | P3 | 3000 |
| Electronics | P4 | 1000 |
| Books | P5 | 2000 |
| Books | P6 | 1500 |
| category | product_id | total_revenue |
| Electronics | P1 | 5000 |
| Electronics | P2 | 3000 |
| Electronics | P3 | 3000 |
| Electronics | P4 | 1000 |
| Books | P5 | 2000 |
| Books | P6 | 1500 |
Interview Tip: Use dense_rank() if you want ties included (both P2 and P3 at rank 2). Use row_number() if you want exactly 3 rows per category regardless of ties. Clarify with the interviewer which behavior they want.
What NOT to Say: "I'd sort each group and take the first 3." Without a window function, you'd need expensive groupBy + collect_list + slice patterns.
Answer First: Within each entity, order observations by time and use lag to retrieve the preceding value. Subtracting it from the current value yields the consecutive change while the first row remains null.
Memory Map: Compute the running difference between consecutive rows -> entity partition orders observations -> lag retrieves the prior value -> subtraction produces consecutive change -> first-row null handling preserves sequence semantics [02_PySpark_Advanced_Operations.md:1712].
Q31: Compute the running difference between consecutive rows.
Simple Explanation:
For time-series data (sensor readings, stock prices), you often want to know: "How much did the value change from the previous reading?" Use lag() to look back one row.
Technical details:
from pyspark.sql.functions import lag, col
from pyspark.sql import Window
w = Window.partitionBy("sensor_id").orderBy("timestamp")
result = df.withColumn("prev_value", lag("value", 1).over(w)) \
.withColumn("delta", col("value") - col("prev_value"))
Sample data flow:
| sensor_id | timestamp | value |
|---|
| S1 | 1:00 PM | 100 |
| S1 | 1:05 PM | 105 |
| S1 | 1:10 PM | 98 |
| sensor_id | timestamp | value |
| S1 | 1:00 PM | 100 |
| S1 | 1:05 PM | 105 |
| S1 | 1:10 PM | 98 |
Interview Tip: lag(col, N) looks N rows back, lead(col, N) looks N rows forward. You can also provide a default value: lag("value", 1, 0) returns 0 instead of NULL for the first row.
What NOT to Say: "I'd do a self-join on row number." Window functions are the standard approach and far more efficient.
Answer First: For each employee, compare their salary against the average for their department. Keep only those who earn more than the average. Window functions let you compute the department average without a separate groupBy + join.
Memory Map: Find employees whose salary is above the department average -> department partition computes mean salary -> original employee rows retain their values -> comparison filters salaries above the window mean -> grouped checks confirm department boundaries [02_PySpark_Advanced_Operations.md:1753].
Q32: Find employees whose salary is above the department average.
Simple Explanation:
For each employee, compare their salary against the average for their department. Keep only those who earn more than the average. Window functions let you compute the department average without a separate groupBy + join.
Technical details:
from pyspark.sql.functions import avg, col
from pyspark.sql import Window
w = Window.partitionBy("department")
result = df.withColumn("dept_avg", avg("salary").over(w)) \
.filter(col("salary") > col("dept_avg")) \
.drop("dept_avg")
Sample data flow:
| name | department | salary |
|---|
| Alice | Eng | 120K |
| Bob | Eng | 80K |
| Charlie | Eng | 100K |
| Diana | Sales | 90K |
| Eve | Sales | 70K |
| name | department | salary |
| Alice | Eng | 120K |
| Bob | Eng | 80K |
| Charlie | Eng | 100K |
| Diana | Sales | 90K |
| Eve | Sales | 70K |
| name | department | salary |
| Alice | Eng | 120K |
| Diana | Sales | 90K |
Interview Tip: This shows the power of window functions over groupBy — you keep individual rows while computing group-level metrics. No self-join needed.
What NOT to Say: "I'd groupBy department, compute the average, then join back." That works but is less elegant and less efficient than a single window function.
Answer First: "Find loyal customers who never return products." This is a classic "exists in A but not in B" pattern. The most efficient approach is a left anti join.
Memory Map: Scenario — Given two DataFrames (orders and returns), find customers who placed orders but never returned anything -> orders establish candidate customers -> return keys mark disqualifying matches -> left anti join retains unmatched purchasers -> distinct-count reconciliation proves exclusion [02_PySpark_Advanced_Operations.md:1805].
Q33: Scenario — Given two DataFrames (orders and returns), find customers who placed orders but never returned anything.
Simple Explanation:
"Find loyal customers who never return products." This is a classic "exists in A but not in B" pattern. The most efficient approach is a left_anti join.
Technical details:
loyal_customers = orders_df.join(returns_df, "customer_id", "left_anti") \
.select("customer_id").distinct()
loyal_customers = orders_df.join(returns_df, "customer_id", "left_outer") \
.filter(returns_df["return_id"].isNull()) \
.select(orders_df["customer_id"]).distinct()
Sample data flow:
| customer_id | order_id | | customer_id | return_id |
|---|
| C1 | O1 | | C1 | R1 |
| C2 | O2 | | C3 | R2 |
| C3 | O3 | | | |
| C4 | O4 | | | |
| C2 | ← has orders, no returns | | | |
| C4 | ← has orders, no returns | | | |
Interview Tip: Always mention left_anti first — it's the most efficient because Spark doesn't need to carry any columns from the right side. It just checks existence. Follow up with the left_outer + filter approach to show you know alternatives.
What NOT to Say: "I'd collect all return customer IDs into a list and use isin()." That collects data to the driver and fails with large datasets.
Answer First: For each product, compare this month's revenue to last month's. Growth rate = (this month - last month) / last month 100%. Use lag() to look back one month.
Memory Map: Calculate month-over-month growth rate per product -> product-month aggregation establishes comparable revenue -> lag retrieves prior month within product -> guarded division computes percentage change -> missing and zero baselines remain explicit [02_PySpark_Advanced_Operations.md:1849].
Q34: Calculate month-over-month growth rate per product.
Simple Explanation:
For each product, compare this month's revenue to last month's. Growth rate = (this month - last month) / last month * 100%. Use lag() to look back one month.
Technical details:
from pyspark.sql.functions import lag, col, round as _round
from pyspark.sql import Window
w = Window.partitionBy("product_id").orderBy("month")
growth = monthly_revenue.withColumn("prev_revenue", lag("revenue", 1).over(w)) \
.withColumn("mom_growth_pct",
_round(
(col("revenue") - col("prev_revenue")) / col("prev_revenue") * 100, 2
)
)
Sample data flow:
| product_id | month | revenue |
|---|
| P1 | 2026-01 | 10000 |
| P1 | 2026-02 | 12000 |
| P1 | 2026-03 | 9000 |
| product_id | month | revenue |
| P1 | 2026-01 | 10000 |
| P1 | 2026-02 | 12000 |
| P1 | 2026-03 | 9000 |
Interview Tip: Handle edge cases: what if prev_revenue is 0? Division by zero! Add a when clause: when(col("prev_revenue") > 0, growth_formula).otherwise(None).
What NOT to Say: "I'd self-join the table on month = month - 1." That works but window functions are the standard, cleaner approach.
Answer First: Given order data (which products were in each order), find which product pairs appear together most often. This is the classic "Customers who bought X also bought Y" problem.
Memory Map: Find all pairs of products frequently bought together (market basket analysis) -> basket groups collect distinct products per order -> pair generation creates canonical combinations -> frequency aggregation counts co-occurrence -> support threshold selects meaningful associations [02_PySpark_Advanced_Operations.md:1895].
Q35: Find all pairs of products frequently bought together (market basket analysis).
Simple Explanation:
Given order data (which products were in each order), find which product pairs appear together most often. This is the classic "Customers who bought X also bought Y" problem.
Technical details:
from pyspark.sql.functions import collect_set, explode, col, array_sort
from itertools import combinations
order_products = df.groupBy("order_id") \
.agg(collect_set("product_id").alias("products"))
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import ArrayType, StructType, StructField, StringType
import pandas as pd
@pandas_udf(ArrayType(StringType()))
def get_pairs(products: pd.Series) -> pd.Series:
return products.apply(lambda x: [f"{a}|{b}" for a, b in combinations(sorted(x), 2)])
pairs = order_products.withColumn("pair", explode(get_pairs(col("products")))) \
.groupBy("pair").count() \
.orderBy(col("count").desc())
Sample data flow:
| order_id | product_id |
|---|
| O1 | Laptop |
| O1 | Mouse |
| O1 | Keyboard |
| O2 | Laptop |
| O2 | Mouse |
| order_id | products |
| O1 | [Keyboard, Laptop, Mouse] |
| O2 | [Laptop, Mouse] |
| Keyboard | Laptop |
| Keyboard | Mouse |
| Laptop | Mouse |
| Laptop | Mouse |
| pair | count |
| Laptop | Mouse |
| Keyboard | Laptop |
| Keyboard | Mouse |
Interview Tip: For very large datasets, consider using Spark's FPGrowth algorithm from MLlib instead of this brute-force approach. Mention it to show ML awareness.
What NOT to Say: "I'd use nested for-loops to compare every pair." That's O(n^2) and not distributed.
Answer First: Real-world data is messy. Some rows might have wrong data types, missing fields, or corrupted characters. Instead of failing the entire pipeline, we separate "good" records from "bad" records and process them differently.
Memory Map: Scenario — Process a large CSV with bad records. Keep good records, quarantine bad ones -> permissive parsing separates malformed input -> corrupt-record column retains rejected payload -> valid rows follow typed processing -> quarantine counts reconcile with source totals [02_PySpark_Advanced_Operations.md:1967].
Q36: Scenario — Process a large CSV with bad records. Keep good records, quarantine bad ones.
Simple Explanation:
Real-world data is messy. Some rows might have wrong data types, missing fields, or corrupted characters. Instead of failing the entire pipeline, we separate "good" records from "bad" records and process them differently.
Analogy: Like airport security — valid passengers go through to their gate, flagged items go to a separate inspection area. You don't shut down the whole airport because of one suspicious bag.
Technical details:
df = spark.read.option("mode", "PERMISSIVE") \
.option("columnNameOfCorruptRecord", "_corrupt_record") \
.schema(expected_schema) \
.csv("/path/to/data.csv")
good_records = df.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
bad_records = df.filter(col("_corrupt_record").isNotNull()) \
.select("_corrupt_record")
good_records.write.format("delta").mode("append").saveAsTable("silver_data")
bad_records.write.format("delta").mode("append").saveAsTable("quarantine_data")
The three read modes:
PERMISSIVE (default): Puts corrupt rows in a special column. Pipeline continues.
DROPMALFORMED: Silently drops bad rows. Dangerous — you lose data without knowing.
FAILFAST: Throws exception on first bad row. Good for testing, bad for production.
Interview Tip: Always mention the quarantine pattern — it shows production maturity. In Databricks, this is part of the Medallion Architecture: bad records go to a quarantine table in the Bronze layer for later investigation.
What NOT to Say: "I'd use FAILFAST mode in production." One bad row would crash your entire pipeline.
Answer First: Spark doesn't have a built-in exact median() function. You have two options: an approximate median (fast, good enough for most cases) or an exact median (uses window functions, more expensive).
Memory Map: a custom aggregation — median (not built into Spark SQL) -> group values enter a mergeable buffer -> partial arrays or sketches combine across partitions -> ordered midpoint yields median estimate -> known distributions validate aggregation [02_PySpark_Advanced_Operations.md:2011].
Simple Explanation:
Spark doesn't have a built-in exact median() function. You have two options: an approximate median (fast, good enough for most cases) or an exact median (uses window functions, more expensive).
Technical details:
from pyspark.sql.functions import percentile_approx, expr
result = df.groupBy("department") \
.agg(percentile_approx("salary", 0.5).alias("median_salary"))
from pyspark.sql.functions import count, row_number, col, avg
from pyspark.sql import Window
w = Window.partitionBy("department").orderBy("salary")
total_w = Window.partitionBy("department")
result = df.withColumn("rn", row_number().over(w)) \
.withColumn("cnt", count("*").over(total_w)) \
.filter(
(col("rn") == (col("cnt") / 2).cast("int") + 1) |
((col("cnt") % 2 == 0) & (col("rn") == (col("cnt") / 2).cast("int")))
) \
.groupBy("department") \
.agg(avg("salary").alias("median_salary"))
Sample data flow:
| salary | rn | cnt |
|---|
| 60K | 1 | 5 |
| 70K | 2 | 5 |
| 80K | 3 | 5 |
| 90K | 4 | 5 |
| 100K | 5 | 5 |
| 60K | 1 | 4 |
| 70K | 2 | 4 |
| 80K | 3 | 4 |
| 90K | 4 | 4 |
Interview Tip: Start with percentile_approx — it's the pragmatic answer. Mention the exact method as a follow-up. If the interviewer asks about accuracy, percentile_approx with default settings is accurate to within 0.01%.
What NOT to Say: "Spark can't compute median." It can — you just need to know the approach.
Answer First: This is a complete end-to-end streaming pipeline. It reads JSON events from Kafka, removes duplicate events (using event id), and writes clean data to a Delta table. This is the bread-and-butter of data engineering on Databricks.
Memory Map: a streaming pipeline that reads from Kafka, deduplicates, and writes to Delta -> Kafka offsets feed streaming input -> event keys and watermark bound deduplication state -> Delta sink commits each batch transactionally -> checkpoint restart proves replay safety [02_PySpark_Advanced_Operations.md:2070].
Q38: Write a streaming pipeline that reads from Kafka, deduplicates, and writes to Delta.
Simple Explanation:
This is a complete end-to-end streaming pipeline. It reads JSON events from Kafka, removes duplicate events (using event_id), and writes clean data to a Delta table. This is the bread-and-butter of data engineering on Databricks.
Technical details:
from pyspark.sql.functions import from_json, col, expr
schema = "event_id STRING, user_id STRING, event_type STRING, event_time TIMESTAMP, payload STRING"
raw = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "broker1:9092") \
.option("subscribe", "events") \
.option("startingOffsets", "latest") \
.load()
parsed = raw.select(
from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")
deduped = parsed \
.withWatermark("event_time", "10 minutes") \
.dropDuplicates(["event_id"])
def upsert_events(batch_df, batch_id):
from delta.tables import DeltaTable
if DeltaTable.isDeltaTable(spark, "/delta/events"):
target = DeltaTable.forPath(spark, "/delta/events")
target.alias("t").merge(
batch_df.alias("s"), "t.event_id = s.event_id"
).whenNotMatchedInsertAll().execute()
else:
batch_df.write.format("delta").save("/delta/events")
deduped.writeStream \
.foreachBatch(upsert_events) \
.option("checkpointLocation", "/checkpoints/events") \
.trigger(processingTime="30 seconds") \
.start()
Interview Tip: This pipeline has TWO levels of deduplication: (1) dropDuplicates removes dupes within the stream, and (2) MERGE prevents dupes against the target table. Mention both — it shows thorough thinking.
What NOT to Say: "I'd just use append mode." Without deduplication, duplicate events from Kafka retries will create duplicate rows in your Delta table.