Apache Spark — Data Engineering Interview Deep Dive

Scope — everything about Apache Spark a data engineering interview probes: the execution model, the APIs and the optimiser, shuffles, partitioning, joins and skew, caching, memory and serialisation, PySpark specifics, writes, Structured Streaming and its watermarks, tuning, debugging and the common failures — answered for Spark 3.5, with Spark 4.0 differences called out. See also: ../faq_DE_interview.md — the high-level map and last-minute checklist across all DE topics; ./stream_pipeline.md — streaming in depth; ./batch_pipeline.md — where a Spark job sits in a pipeline; ./hadoop.md — HDFS, YARN and MapReduce; ../spark/faq_spark_hadoop.md — the question bank: short spoken answers, troubleshooting walk-throughs, pipeline-design prompts and a study plan.

Every answer leads with the sentence you would say out loud, then the mechanism, then the trade-off the interviewer pushes on next. Configuration names and defaults are Spark 3.5’s unless a section says otherwise.

0) How to Use This Doc with the Question Bank

This is the deep dive: the mechanism behind each answer. The question bank, faq_spark_hadoop.md, holds the short spoken answers, the drills and the study plan, and its review order lines the two files up step by step.

• Follow that review order, not §1 → §16. This file’s sections follow how Spark is built, which is not the order an interview probes it.
• Come here when a question-bank answer does not survive “why?”. Each of its sections links to the matching section here.
• §10 (PySpark) and §11 (writes and table formats) have no counterpart in the question bank. Read them for a data-engineering loop.
• §16-1’s quick-fire answers are a recap. The full mock bank, with difficulty levels, is the question bank’s §14-2.

1) What Spark Is, and Where It Fits ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

1-1) The one-paragraph answer

Spark is a distributed, in-memory-first compute engine: you describe a computation over a dataset, Spark turns it into a plan of parallel tasks, and runs those tasks on a cluster, recomputing lost pieces from their lineage.

• Unified engine — batch (Spark SQL / DataFrames), streaming (Structured Streaming), machine learning (MLlib) and graph processing share one engine and one API surface.
• Storage-agnostic — reads and writes HDFS, S3 / GCS / ADLS, JDBC, Kafka, Delta, Iceberg, Hudi, Parquet, ORC, Avro, JSON, CSV.
• Languages — Scala, Java, Python (PySpark), SQL and R; Spark Connect (3.4+) adds a thin client that talks to a remote driver over gRPC.

1-2) Spark vs MapReduce

Spark keeps intermediate results in memory and plans a whole DAG of stages; MapReduce writes to HDFS between every job, so multi-step and iterative work is much slower on it.

Spark still shuffles through local disk, so “in-memory” does not mean “never touches disk”.

1-3) Spark vs the alternatives

The honest interview line: “if it fits on one machine, don’t use Spark.”

2) Architecture and the Execution Model ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

2-1) Driver, cluster manager, executors

One driver plans the job, many executors run it in parallel, and a cluster manager hands out the machines.

            ┌──────────── Driver ────────────┐
            │ SparkSession → DAG → Stages     │
            │ DAGScheduler / TaskScheduler    │
            └──────┬──────────────────┬───────┘
                   │ request resources │
            ┌──────▼──────┐            │ send tasks
            │Cluster Mgr  │            │
            │YARN / K8s   │            │
            └──────┬──────┘            │
        ┌──────────┴───────┬───────────▼──────┐
   ┌────▼─────┐      ┌─────▼────┐       ┌─────▼────┐
   │Executor 1│      │Executor 2│  ...  │Executor N│
   │task task │      │task task │       │task task │
   │ cache    │      │ cache    │       │ cache    │
   └──────────┘      └──────────┘       └──────────┘

• Concurrent task slots = executors × spark.executor.cores. A 200-partition stage on 50 slots (say 10 executors × 5 cores) runs in 4 waves; actual throughput still depends on the workload and I/O.
• The driver is a single point of failure for the application: if it dies, the job dies (YARN can retry the whole application with spark.yarn.maxAppAttempts).

2-2) Deploy modes and spark-submit

In client mode the driver runs where you submitted from; in cluster mode it runs inside the cluster — production jobs use cluster, so a laptop disconnect cannot kill them.

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 10 \
  --executor-cores 5 \
  --executor-memory 16g \
  --driver-memory 4g \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.sql.adaptive.enabled=true \
  etl_daily_orders.py --run-date 2026-11-01

2-3) Job → stage → task, and the two schedulers

An action creates a job; the DAGScheduler cuts the job into stages at shuffle boundaries; the TaskScheduler sends one task per partition of each stage to an executor.

Job   = everything one action needs
Stage = a run of narrow transformations between two shuffles
Task  = one partition of one stage, run on one core

df.read ─ filter ─ withColumn │ shuffle │ groupBy().agg ─ write
└──────── Stage 0 ───────────┘         └──── Stage 1 ─────┘
  200 input partitions → 200 tasks        shuffle partitions → N tasks

• DAGScheduler — builds stages, tracks which shuffle outputs exist, resubmits a stage whose shuffle files were lost.
• TaskScheduler — places tasks, preferring data locality (PROCESS_LOCAL → NODE_LOCAL → RACK_LOCAL → ANY, waiting up to spark.locality.wait, default 3s, at each level), and retries a failed task up to spark.task.maxFailures (default 4).
• Stages with no dependency on each other can run at the same time.

2-4) Fault tolerance: lineage, retries and speculation

Spark does not replicate computed data; it remembers how each partition was built and recomputes a lost one from its parents.

• Lineage — every RDD / DataFrame knows its parent and the function applied. Lose an executor and its cached partitions are recomputed; lose its shuffle files and the map stage that produced them is re-run (unless an external shuffle service still serves them).
• Task retry — a failed task is retried on another executor; after spark.task.maxFailures failures the stage, and the job, fails.
• Speculation (spark.speculation=true) — launches a duplicate of an unusually slow task and takes whichever finishes first. It helps with a bad node; it does nothing for skew, because the duplicate has the same oversized partition. It also means side effects inside tasks can run twice.
• Checkpointing (df.checkpoint() with spark.sparkContext.setCheckpointDir(...)) writes data to reliable storage and cuts the lineage — useful for long iterative plans.

2-5) Dynamic allocation

Dynamic allocation adds executors when tasks queue up and releases idle ones, so one job does not hold a fixed fleet for its whole run.

• spark.dynamicAllocation.enabled=true, bounded by minExecutors / maxExecutors.
• Releasing an executor would lose its shuffle files, so it needs either an external shuffle service (YARN) or spark.dynamicAllocation.shuffleTracking.enabled=true (keeps executors holding live shuffle data, the usual choice on Kubernetes).
• Trade-off: better cluster utilisation against slower ramp-up and executors churned between stages.

3) APIs: RDD, DataFrame, Dataset ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

3-1) The comparison

Use DataFrames: they carry a schema, so Catalyst can optimise them and Tungsten can store them compactly; an RDD is a collection of opaque objects run through functions Spark cannot see into.

In Python and SQL a DataFrame is all there is; in Scala, DataFrame is literally Dataset[Row].

3-2) Transformations vs actions, and lazy evaluation

Transformations are lazy and only add to a plan; an action runs the plan — which is what lets Spark optimise the whole thing before doing any work.

# python — nothing runs until the last line
from pyspark.sql import functions as F

orders = spark.read.parquet("s3://lake/orders/")            # reads only the footer / schema
big = orders.filter(F.col("amount") > 100)                  # lazy
by_day = big.groupBy("order_date").agg(F.sum("amount"))     # lazy
by_day.write.mode("overwrite").parquet("s3://lake/agg/")    # action: the job runs here

Consequence people trip on: calling two actions on the same un-cached DataFrame computes it twice from the source.

3-3) Narrow vs wide dependencies

A narrow transformation needs only one input partition per output partition and stays in the same stage; a wide one needs data from many partitions, so it forces a shuffle and starts a new stage.

3-4) RDD-era questions that still get asked

reduceByKey combines on the map side before the shuffle; groupByKey ships every value across the network and then groups — so reduceByKey (or a DataFrame groupBy().agg) is the right choice for an aggregation.

# python — word count, the RDD way
lines = spark.sparkContext.textFile("s3://lake/raw/books/*.txt")
counts = (lines.flatMap(lambda line: line.split())
               .map(lambda word: (word, 1))
               .reduceByKey(lambda a, b: a + b))      # map-side combine, then shuffle
top = counts.takeOrdered(10, key=lambda kv: -kv[1])

• map vs flatMap — one output per input vs zero-or-more outputs per input.
• mapPartitions — runs once per partition; open one DB connection per partition, not per row.
• Broadcast variable — a read-only value shipped once per executor (a lookup dict).
• Accumulator — a write-only counter aggregated on the driver. Updates inside actions are applied exactly once; updates inside transformations can be applied more than once when a task or stage is re-run, so do not use them for exact counts there.

4) Catalyst, Tungsten and AQE ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

4-1) Catalyst: how a query becomes a plan

Catalyst turns your DataFrame or SQL into a logical plan, rewrites it with rules, picks a physical plan, and generates code for it.

SQL / DataFrame
   │ parse
   ▼
Unresolved logical plan ── analysis (catalog: tables, columns, types) ──► Analysed plan
   │ rule-based optimisation
   ▼
Optimised logical plan   (predicate pushdown, column pruning, constant folding,
   │                      join reordering with CBO stats, subquery rewrites)
   │ physical planning (strategies: which join, which aggregate)
   ▼
Physical plan  ──► whole-stage code generation ──► RDD of InternalRow, run on executors

• Predicate pushdown — a filter moves down to the scan, so Parquet / ORC skip row groups by their min / max statistics, and JDBC sources push it into the WHERE.
• Column pruning — only the referenced columns are read from a columnar file.
• Partition pruning — a filter on a partition column skips whole directories; dynamic partition pruning (3.0+) does it at runtime from the other side of a join.
• CBO (spark.sql.cbo.enabled) uses table statistics from ANALYZE TABLE ... COMPUTE STATISTICS for join ordering; without stats, Spark estimates from file sizes.

4-2) Tungsten: the execution engine

Tungsten stores rows in a compact binary format, manages memory itself, and fuses operators into generated Java code, so a stage runs like one hand-written loop.

• UnsafeRow — binary row format, off the JVM object model: less memory, almost no GC pressure.
• Whole-stage code generation — filter → project → aggregate become one generated function; in explain() output these operators are marked with *(n).
• Cache-aware sorting and hashing for aggregations and joins.
• What defeats it: Python UDFs, Scala lambdas on Datasets, and very wide schemas (code generation is skipped above spark.sql.codegen.maxFields).

4-3) Adaptive Query Execution (AQE)

AQE re-optimises the plan between stages using the real sizes of the shuffle output, instead of trusting estimates made before the job started. It is on by default since Spark 3.2.

• AQE only acts at shuffle boundaries — it cannot fix a plan with no exchange.
• By default a partition counts as skewed when it is larger than both skewedPartitionFactor × the median partition size and skewedPartitionThresholdInBytes (256 MB).

4-4) Reading a physical plan

df.explain("formatted") shows the physical plan: read it bottom-up, and look for the scan filters, every Exchange (a shuffle), and the join operator.

# python
(spark.table("sales").filter("sale_date = '2026-11-01'")
      .join(spark.table("stores"), "store_id")
      .groupBy("region").sum("amount")
      .explain(mode="formatted"))

explain modes: simple, extended (all four plans), codegen, cost, formatted.

5) Shuffles ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

5-1) What a shuffle actually does

A shuffle redistributes rows so that all rows with the same key land in the same partition: every map task writes its output to local disk sorted by target partition, and every reduce task fetches its slice from every map task over the network.

Map side (stage N)                        Reduce side (stage N+1)
task 0: rows → partition by hash(key)     task 0 fetches block 0 from every map output
        → sort / spill → data file        task 1 fetches block 1 from every map output
          + index file (one per task)     ...
task 1: ...                               then aggregate / join the fetched rows
        M map tasks × R reduce partitions = up to M × R blocks fetched

• Spark uses the sort-based shuffle: one data file and one index file per map task.
• Cost: serialisation, disk write, network transfer, disk read, deserialisation — usually the most expensive part of a job, and where most spills and OOMs happen. §9-5 follows one row through it.
• External shuffle service keeps serving shuffle files after the executor that wrote them is gone (needed for dynamic allocation on YARN).

5-2) How to avoid or shrink a shuffle

• Filter and project early so less data reaches the exchange (Catalyst does much of this; a UDF in the way stops it).
• Broadcast the small side of a join.
• Pre-aggregate — groupBy().agg already combines on the map side; avoid collect_list then aggregate.
• Bucket tables written repeatedly joined on the same key (section 6-4).
• Do not repartition “just in case” — every repartition is a full shuffle.
• Compress shuffle data (spark.shuffle.compress, on by default) and serialise with Kryo for RDD workloads. Kryo does not change a DataFrame shuffle (§9-4).

5-3) spark.sql.shuffle.partitions

It sets how many partitions every DataFrame shuffle produces — 200 by default, which is wrong for both tiny and huge jobs.

Rule of thumb: total shuffle size ÷ ~128 MB, and at least 2–3× the number of cores. With AQE on, set it generously high and let coalescing merge the small partitions.

6) Partitioning and File Layout ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

6-1) Partitions in memory

A partition is the chunk of data one task processes; the number of partitions is the parallelism of a stage.

• On read, file sources split input into partitions of up to spark.sql.files.maxPartitionBytes (default 128 MB); a non-splittable file (gzip CSV) is one partition no matter its size.
• After a shuffle, the count is spark.sql.shuffle.partitions (or AQE’s coalesced count).
• df.rdd.getNumPartitions() tells you the current count.

6-2) repartition vs coalesce

repartition(n) does a full shuffle and can go up or down; coalesce(n) only merges existing partitions without a shuffle, so it can only go down.

repartitionByRange(n, col) range-partitions by sampling — useful before writing sorted output.

6-3) partitionBy on write, and the small-files problem

df.write.partitionBy("dt") is a directory layout for pruning on read, not the same thing as in-memory partitions — and every in-memory partition can write one file into every directory value it holds.

# python — one tidy set of files per date: shuffle by the partition column first
(df.repartition("event_date")
   .write.mode("overwrite")
   .partitionBy("event_date")
   .parquet("s3://lake/events/"))

• Without the repartition, 200 tasks × 30 dates can write 6,000 small files.
• Small files hurt every reader: one task per file, footer reads, slow object-store listings, NameNode memory on HDFS.
• Fixes: coalesce / repartition before the write, AQE coalescing, a compaction job (Delta OPTIMIZE, Iceberg rewrite_data_files). maxRecordsPerFile only caps a file’s size; it never merges files.
• Partition by a low-cardinality column queries filter on (a date). Partitioning by user_id creates millions of directories.

6-4) Bucketing

When a join runs every day on the same key, bucket both tables on that key with the same bucket count. Spark then reads bucket i of each table in the same task and drops the Exchange on both sides of the join. Show that with EXPLAIN FORMATTED; having the DDL is not evidence.

The cost this removes is the run-time hash partitioning of both sides (§5-1): serialise, write, send, read and deserialise, on every run. Bucketing pays a similar cost once, when the table is written, and every later join reuses the layout.

Writing the tables

-- spark sql — CLUSTERED BY ... INTO n BUCKETS is bucketing, not clustering
CREATE TABLE customers_bucketed
USING PARQUET
CLUSTERED BY (customer_id) SORTED BY (customer_id) INTO 64 BUCKETS
AS SELECT /*+ REPARTITION(64, customer_id) */ customer_id, customer_name
FROM customers;

CREATE TABLE orders_bucketed
USING PARQUET
PARTITIONED BY (order_date)                 -- prune by day, bucket within each day
CLUSTERED BY (customer_id) SORTED BY (customer_id) INTO 64 BUCKETS
AS SELECT /*+ REPARTITION(64, customer_id) */ order_id, customer_id, amount, order_date
FROM orders;

# python — the DataFrame form; bucketBy needs saveAsTable, because the layout lives in the catalog
(orders.repartition(64, "customer_id")      # task i holds exactly bucket i: one file per bucket
       .write.bucketBy(64, "customer_id").sortBy("customer_id")
       .mode("overwrite").saveAsTable("analytics.orders_bucketed"))

• Repartition to the bucket count before the write. Without it, every write task writes a file into every bucket. In a local test, 4 tasks and 64 buckets wrote 256 files; with the repartition, 64. A repartition into the same number of partitions on the same column puts each bucket in a single task, because both use the same hash.
• Read it as a table, with spark.table(...) or the SQL name. spark.read.parquet(path) sees plain files, because the bucket spec is stored in the metastore, not in the files.
• Partitioned and bucketed together works. Each order_date directory holds its own 64 buckets, and the join reads bucket i from every selected day in task i, so pruning to two days still joins with no shuffle.
• Delta Lake tables do not support bucketBy. On a lakehouse the counterpart is Iceberg’s bucket(n, col) partition transform (see the comparison table below).

What has to match, or the Exchange comes back

Each of these was checked on Spark 4.0.1, with broadcast joins disabled so a sort-merge join was forced:

spark.sql.sources.bucketing.enabled must stay true (the default). With spark.sql.sources.bucketing.autoBucketedScan.enabled (also on by default), Spark switches a bucketed scan off when the query cannot use it. That is the disabled by query planner text.

Proving it with EXPLAIN FORMATTED

-- before: plain tables — both sides are hash-partitioned at run time
SortMergeJoin Inner
:- Sort
:  +- Exchange hashpartitioning(customer_id, 200)
:     +- Scan parquet customers
+- Sort
   +- Exchange hashpartitioning(customer_id, 200)
      +- Scan parquet orders

-- after: both bucketed 64 / 64 on customer_id — nothing moves before the join
HashAggregate                                  <- GROUP BY customer_id, customer_name: no Exchange either
+- HashAggregate
   +- SortMergeJoin Inner
      :- Sort
      :  +- Scan parquet customers_bucketed    Bucketed: true, SelectedBucketsCount: 64 out of 64
      +- Sort
         +- Scan parquet orders_bucketed       Bucketed: true, SelectedBucketsCount: 64 out of 64

• The claim is “no Exchange between the join and either scan.” Follow both input branches down through the Sort to the scan. In the plan before bucketing, too, a Sort is the join’s direct child and the Exchange sits under it. Exchanges above the join are separate questions.
• Bucketed: true on both scans is not proof by itself. Bucketed: false (disabled by query planner) only says that side’s buckets are unused: a broadcast join’s streamed side reports it without being shuffled. With autoBucketedScan.enabled=false, a scan still reports Bucketed: true when an Exchange sits right above it, so the branches decide.
• The Sort stays. By default Spark 3+ does not rely on the per-bucket sort order, because a bucket can be several files. So SORTED BY is not what removes the shuffle.
• A GROUP BY that contains the bucket key reuses the join’s partitioning, so the plan above has no Exchange after the join either. GROUP BY order_date instead adds an Exchange above the join, because the rows have to be regrouped by a different key.

Bucketing vs clustering: the same word for different things

Clustering makes a filter read less, and bucketing makes a join move less. Neither one does the other’s job. Physical clustering keys per engine are covered in ./schema_design.md.

Choosing the bucket count, and keeping the layout

• A bucketed join runs exactly n tasks. With 64 buckets over 2 TB of orders, each task gets 32 GB, which is far too much, and every core beyond the 64th sits idle. Size n so that a bucket of the larger table is a sensible task (hundreds of MB to a couple of GB). For example, 1,024 buckets gives about 2 GB of orders and about 100 MB of a 100 GB customers table per task. With daily partitions, size n from the days that one join actually reads.
• Use a power of two. That keeps bucket coalescing available when a table later doubles.
• Changing n means rewriting the whole table, so size it for the growth you expect.
• Appends keep the table bucketed, but they add files. In the test, one small INSERT INTO wrote one more file per bucket per write task (64 files became 320). Compact the table periodically by rewriting it with the same REPARTITION, or make each daily partition a single write.
• It only pays for itself on repeated joins. For a join that runs once, the write cost is roughly the same as the shuffle it saves.

What bucketing does not fix: skew

A hot customer_id still hashes to one bucket, so one task does most of the work. A bucketed join is also worse to skew than a shuffled one: AQE’s skew-join handling splits shuffle partitions (§4-3), and a bucketed join has none, so AQE has nothing to split. You can handle the hot keys separately (filter them out and broadcast-join them), or turn bucketing off for that one query (spark.sql.sources.bucketing.enabled=false) so that AQE can split it. See §7-3.

Claims to correct when you hear them

• “Network cost drops to zero.” What goes away is the shuffle. Bucket files on S3 or another object store are still read over the network, and Spark does not guarantee that bucket i of both tables sits on one node. Reading both bucket i files in one task is enough.
• “Success looks like a BroadcastHashJoin with no Exchange.” A broadcast has its own BroadcastExchange, and when it happens Spark does not use the buckets.
• “A different bucket count turns the optimisation off.” Only the side with fewer buckets is shuffled. With coalescing enabled, in a sort-merge join, neither side is, provided one count divides the other within maxBucketRatio: 64 against 128 coalesces, and 64 against 100 does not.
• “Set autoBroadcastJoinThreshold = -1 to make bucketing work.” That only forces a sort-merge join on small demo tables. In production, a side small enough to broadcast should be broadcast.

Saying it out loud

For a join that runs every day on customer_id, I would bucket both tables by customer_id with the same power-of-two bucket count, written with a repartition so each bucket is one file. Then Spark joins bucket i to bucket i without shuffling either side. I would confirm it in EXPLAIN FORMATTED: no Exchange on either branch of the join, and Bucketed: true on both scans. It costs extra at write time and the bucket count is fixed, so it is only worth doing for a recurring join. It also makes skew harder, because AQE cannot split a bucket. On Delta, liquid clustering is not a substitute. It improves file skipping but still shuffles, and the lakehouse equivalent is an Iceberg bucket transform with a storage-partitioned join.

7) Joins and Data Skew ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

7-1) The join strategies

Spark picks a broadcast hash join when one side is small enough, and a sort-merge join otherwise; everything else is a special case.

# python — hint the strategy
from pyspark.sql.functions import broadcast

joined = facts.join(broadcast(dim_store), "store_id")

-- Spark SQL join hints (3.0+): BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL
SELECT /*+ BROADCAST(s) */ f.*, s.region
FROM   sales f
JOIN   stores s ON f.store_id = s.store_id;

Broadcasting something too large moves the OOM from the executors to the driver — the threshold exists for a reason.

7-2) Join types and their traps

• NULL keys never match in an equi-join; use eqNullSafe (<=> in SQL) if they must.
• Joining on columns with the same name via a list (join(df2, "id")) avoids the ambiguous duplicate id column you get with df1.id == df2.id.

7-3) Data skew: spotting it and fixing it

Skew is a few keys holding most of the rows, so one task runs for an hour while the rest finish in a minute — total job time is set by the slowest task.

How to spot it: in the Spark UI stage page, the max task duration and shuffle-read size are many times the median, and one task is still running long after the others.

How to fix it, in the order you would try:

1. AQE skew join — spark.sql.adaptive.skewJoin.enabled=true (default with AQE) splits oversized partitions of a sort-merge join automatically.
2. Broadcast the small side, so the big side never shuffles by the key.
3. Handle the hot key separately — it is often NULL, -1, "unknown" or a test account: filter it out, join the rest normally, process the hot key with a broadcast, union.
4. Salting — append a random suffix to the key on the big side and replicate the small side once per suffix.
5. Two-stage aggregation for a skewed groupBy — aggregate by (key, salt), then by key.

# python — salting a skewed join
from pyspark.sql import functions as F

SALT = 16
big_salted = big.withColumn("salt", (F.rand() * SALT).cast("int"))
small_salted = small.crossJoin(spark.range(SALT).withColumnRenamed("id", "salt"))
joined = big_salted.join(small_salted, ["key", "salt"]).drop("salt")

# python — two-stage aggregation for a skewed groupBy
partial = (events.withColumn("salt", (F.rand() * 16).cast("int"))
                 .groupBy("key", "salt").agg(F.sum("amount").alias("part")))
result = partial.groupBy("key").agg(F.sum("part").alias("amount"))

Salting only works for aggregations that can be split and recombined (sum, count, max) — not for an exact countDistinct without extra care.

8) Caching and Persistence ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

8-1) When to cache

Cache a DataFrame only when more than one action reuses it and recomputing it is expensive; otherwise caching costs memory and buys nothing.

• Good: a cleaned dataset used by several aggregations; an iterative ML loop; a dimension joined many times in one job.
• Bad: a DataFrame used once; a huge dataset that evicts everything else; caching “for speed” before a single write.
• cache() is lazy — the data is materialised by the next action; count() is the usual way to force it. Always unpersist() when done.

8-2) Storage levels

• RDD.cache() = MEMORY_ONLY; DataFrame.cache() = MEMORY_AND_DISK (stored in Tungsten’s columnar in-memory format).
• Cached data lives on executors — losing one recomputes its partitions from lineage.

8-3) Cache vs checkpoint vs writing to storage

9) Memory, Serialisation and Resource Sizing ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

9-1) Unified memory management

Executor heap is split into a reserved slice, user memory, and a unified region that execution (shuffles, joins, sorts, aggregations) and storage (cache, broadcasts) share — execution can evict cached blocks, storage cannot evict execution.

Executor container (YARN / K8s pod)
├── spark.executor.memoryOverhead     off-heap: JVM metaspace, thread stacks, NIO buffers,
│                                     Python workers (default max(384 MB, 10% of executor memory))
└── spark.executor.memory (JVM heap)
    ├── Reserved                      300 MB
    ├── User memory                   (1 − spark.memory.fraction) of the rest — your objects, UDF data
    └── Unified memory                spark.memory.fraction = 0.6 of (heap − 300 MB)
        ├── Storage                   spark.memory.storageFraction = 0.5 is protected from eviction
        └── Execution                 can borrow all free storage, and evict cache above the protected part

• When execution memory runs out, operators spill to disk (slow but survives); an OOM happens when something cannot spill — a huge single record, a shuffle-hash build side, a Python worker, or overhead exhausted.
• The container is killed by YARN / Kubernetes when heap + overhead exceed its limit — “Container killed for exceeding memory limits” means overhead, not heap.

9-2) Sizing executors

Use mid-sized executors — around 4–5 cores and enough memory for each core’s share of a partition — rather than one huge executor per node or one core per executor.

• Leave room on each node for the OS and the NodeManager / kubelet.
• PySpark: Python workers live outside the heap, so raise spark.executor.memoryOverhead or set spark.executor.pyspark.memory explicitly.
• Driver memory matters only for what comes back to it: collect, toPandas, broadcasts, and very large plans / file listings.

9-3) Serialisation and deserialisation: what and where

Serialisation turns an in-memory object into bytes that can cross a process boundary or sit on disk; deserialisation rebuilds the object from those bytes. Spark does it every time data leaves an executor’s heap — shipping tasks, shuffles, serialised caching, broadcasts, results and the Python boundary — so every one of those paths pays CPU for it.

An object on the JVM heap is a graph of headers and pointers into one process’s memory. Another executor, another machine or a file cannot follow those pointers, so the data has to be flattened into a self-contained byte stream first.

executor A heap               on the wire / on disk             executor B heap
("U1", 100)   ── serialise ──▶  0a 02 55 31 64 …  ── deserialise ──▶  ("U1", 100)
object graph: headers,          flat, self-contained,               a new object graph,
pointers, boxed fields          compressible                        rebuilt field by field

Closures are serialised too, so a lambda that captures a non-serialisable object (a DB connection, a logger, the enclosing class) fails with Task not serializable before a single task runs. Create the object on the executor instead, and make the write idempotent: a retried or speculative task runs insert again (§15-3).

// scala — fails: `conn` is opened on the driver and captured by the closure
val conn = DriverManager.getConnection(url)
rdd.foreach(row => insert(conn, row))            // Task not serializable

// works: the connection is opened on the executor, once per partition
rdd.foreachPartition { rows =>
  val conn = DriverManager.getConnection(url)
  try rows.foreach(row => insert(conn, row)) finally conn.close()
}

9-4) The serialisers: Java, Kryo and Tungsten’s binary rows

spark.serializer chooses between Java serialisation (the default) and Kryo for RDD records, broadcasts and results. DataFrame rows go through neither: Tungsten keeps them as UnsafeRow bytes from the start, so a SQL shuffle has nothing to convert.

spark.serializer is a core setting: set it before the SparkContext starts (builder, spark-submit --conf or spark-defaults.conf). On Spark 3, spark.conf.set("spark.serializer", …) on a running session is rejected.

# python
spark = (SparkSession.builder
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.classesToRegister", "com.acme.Click,com.acme.Session")
    .config("spark.kryo.registrationRequired", "true")  # fail fast on an unregistered class
    .config("spark.kryoserializer.buffer.max", "256m")  # default 64m
    .getOrCreate())

Traps

• Kryo does not speed up a DataFrame shuffle. SQL exchanges use UnsafeRowSerializer whatever spark.serializer says. Kryo pays off for RDD code and broadcasts of JVM objects.
• An unregistered class still works, slowly. Unless registrationRequired is on, Kryo writes the full class name with every object, which is the overhead you switched to Kryo to remove.
• Buffer overflow. Available: 0, required: … means one object outgrew spark.kryoserializer.buffer.max; raise it (it must stay under 2048m).
• Typed Dataset lambdas convert every row. ds.map(o => …) deserialises each UnsafeRow into a JVM object and serialises the result back — DeserializeToObject and SerializeFromObject in the plan. Column expressions skip both.
• Spark already uses Kryo internally to shuffle RDDs of primitive types, arrays of them and strings (since Spark 2.0), so switching helps most with RDDs of your own classes.

9-5) A shuffle, byte by byte

A row that crosses a shuffle is serialised on the map side, buffered, compressed and written to a local shuffle file, fetched over the network by the reduce task, then decompressed and deserialised — CPU twice, disk twice and the network once, for every row.

# python
orders = spark.createDataFrame(
    [("U1", 100), ("U2", 200), ("U1", 300), ("U3", 400), ("U2", 500), ("U1", 600)],
    ["user_id", "amount"])
orders.groupBy("user_id").sum("amount").show()   # U1 1000, U2 700, U3 400

illustrative layout: 3 map tasks, 3 shuffle partitions

map side: partial sums first                 reduce side: hash(user_id) mod 3
task 0: (U1,100) (U2,200) → U1:100 U2:200    p0 ◀ U1:100 U1:300 U1:600 → U1 1000
task 1: (U1,300) (U3,400) → U1:300 U3:400    p1 ◀ U2:200 U2:500        → U2  700
task 2: (U2,500) (U1,600) → U2:500 U1:600    p2 ◀ U3:400               → U3  400

map task                                     reduce task
1. compute the row, pick its partition id    5. fetch its block from every map output:
2. serialise it into the writer's buffers      network, or a local read when co-located
3. group the bytes by partition id: with        (up to spark.reducer.maxSizeInFlight = 48m)
   ≤ 200 partitions, as here, the bypass     6. decompress and deserialise
   writer streams each to its own file, no   7. aggregate, join or sort
   sort; the sort writers sort and spill
   sorted runs when memory runs out
4. end with one data file + one index file,
   compressed with lz4, through a 32k
   spark.shuffle.file.buffer per stream

The diagram’s 3 × 3 layout is for illustration. A real run splits the list by the session’s default parallelism, shuffles into spark.sql.shuffle.partitions (200) partitions, and AQE may coalesce those. Both 3 and 200 are within the bypass threshold, so the writer is the same.

The partial aggregate (HashAggregate before the Exchange, §4-4) is what shrinks the shuffle. It is an operator of its own, so the shuffle under it has no map-side combine. Here every map task holds distinct keys, so nothing combines; with millions of rows per key it sends one partial sum per key per task instead of every row.

How the map side sorts depends on which writer Spark picks:

The serialised shuffle sorts by partition id only, not by key. Ordering by key, for a sort-merge join, is a separate Sort on the reduce side.

The shuffle files stay on the map executor’s local disk. The external shuffle service keeps serving them after that executor is removed (§5-1); shuffle tracking instead keeps the executor alive while its output is still needed (§2-5). Without either, a lost output is recomputed by rerunning its map tasks.

9-6) Cutting serialisation cost

First move fewer bytes, then make each byte cheaper: shrink what reaches the exchange, keep rows in DataFrame operators so they stay UnsafeRow, and only then tune the serialiser and buffers.

• Catalyst already pushes a filter written after a join below it, so reordering the code often changes nothing. Check explain("formatted") before claiming a win.
• spark.shuffle.file.buffer is per output stream, not per executor, and the bypass writer opens one stream per reduce partition. Many streams × a large buffer is real memory, so raise it only when shuffle write time is the bottleneck, and watch GC.
• A high Task Deserialization Time, or the driver warning “Broadcasting large task binary”, points at an oversized task: the warning measures the stage’s serialised RDD and function, and the metric also counts loading dependencies and a task’s own partition data. Look for a large captured value first, and broadcast it if that is the cause. A different warning, “Stage N contains a task of very large size”, is about a single serialised task, usually one carrying local data from parallelize.
• In an interview, name the bottleneck from the Spark UI — serialisation CPU, fetch wait, spill or network — before naming a fix. Adding executors helps none of the four by itself.

10) PySpark Specifics ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

10-1) How PySpark runs

PySpark is a Python API over the JVM engine: DataFrame operations are planned and run in the JVM, and only Python UDFs and RDD lambdas actually run Python code — in separate Python worker processes on each executor.

Python driver ──Py4J──► JVM driver (Catalyst, scheduler)
                              │ tasks
                              ▼
                        Executor JVM ──(rows, pickled or Arrow)──► Python worker(s)
                                                                    run your UDF / lambda

So a pipeline written only with built-in functions runs at JVM speed from Python.

10-2) UDFs vs pandas UDFs vs built-ins

Prefer built-in functions; if you need custom logic, use a vectorised pandas_udf; a plain Python UDF is the slowest option because it moves rows one at a time.

# python — Spark 3.x pandas UDF with type hints
import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def fahrenheit_to_celsius(f: pd.Series) -> pd.Series:
    return (f - 32) * 5.0 / 9.0

df = df.withColumn("temp_c", fahrenheit_to_celsius("temp_f"))

• A UDF also blocks predicate pushdown through it.
• Grouped logic: groupBy(...).applyInPandas(func, schema) hands each group to func as one pandas DataFrame — a large group must fit in one worker’s memory.
• Spark 3.5 also has Arrow-optimised Python UDFs (useArrow=True), which batch the transfer.

10-3) Arrow, toPandas and collect

• spark.sql.execution.arrow.pyspark.enabled=true makes toPandas() and createDataFrame(pandas_df) transfer columnar batches instead of pickled rows.
• toPandas() and collect() bring all the data to the driver — fine for a summary, an OOM for a table. Use limit(), aggregate first, or write to storage.
• pyspark.pandas (formerly Koalas) gives a pandas-like API that executes on Spark.

11) Spark SQL, Writes and Table Formats ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

11-1) Spark SQL and the catalog

Spark SQL and the DataFrame API compile to the same plans; choose by readability, not speed.

-- Spark SQL: register and query a view
CREATE OR REPLACE TEMP VIEW orders AS
SELECT * FROM parquet.`s3://lake/orders/`;

SELECT customer_id,
       SUM(amount)                                                        AS total,
       ROW_NUMBER() OVER (PARTITION BY country ORDER BY SUM(amount) DESC) AS rank_in_country
FROM   orders
WHERE  order_date >= DATE '2026-10-01'
GROUP  BY customer_id, country;

• Temp views live in one session; global temp views in global_temp across sessions in one app; tables live in the catalog (Hive Metastore, AWS Glue, Unity Catalog, an Iceberg REST catalog).
• Spark 4.0 turns ANSI SQL mode on by default (spark.sql.ansi.enabled=true): invalid casts and integer overflow raise errors instead of returning NULL — a common upgrade surprise.

11-2) Save modes and dynamic partition overwrite

overwrite on a partitioned table replaces the whole table by default; set partitionOverwriteMode=dynamic so it replaces only the partitions present in the data you write — the setting that makes a daily job safely re-runnable.

# python — idempotent daily load: replaces only event_date = run_date
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

(daily_df.write.mode("overwrite")
         .partitionBy("event_date")
         .parquet("s3://lake/events/"))

• INSERT OVERWRITE TABLE t PARTITION (dt = '2026-11-01') SELECT ... is the SQL equivalent for Hive-format tables.
• Plain Parquet directories have no transactions: a failed overwrite can leave a partition half-written, and readers can see it mid-write. Table formats fix this.

11-3) Delta Lake / Iceberg / Hudi with Spark

A table format adds a transaction log over Parquet files, which gives Spark atomic commits, MERGE, time travel and schema evolution on object storage.

-- Spark SQL on Delta Lake: upsert a batch of changes
MERGE INTO dim_customer AS t
USING updates AS s
ON    t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

See ./batch_pipeline.md for how they fit a pipeline and ./schema_design.md for layout and evolution.

11-4) Reading data well

• Supply a schema for JSON / CSV instead of inferSchema — inference costs an extra pass and can guess wrong types.
• mode for malformed records: PERMISSIVE (default, puts bad rows in _corrupt_record), DROPMALFORMED, FAILFAST.
• JDBC reads are single-threaded unless you set partitionColumn, lowerBound, upperBound and numPartitions — and those bounds only decide the stride, they do not filter rows.
• mergeSchema on Parquet reads reconciles differing file schemas, at a cost; avoid it as a default.

12) Structured Streaming and Watermarks ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

12-1) The model

Structured Streaming treats a stream as an unbounded table and runs your DataFrame query on it incrementally, by default as a series of micro-batches, with progress tracked in a checkpoint directory.

# python — Kafka → windowed counts → Delta, Spark 3.5
from pyspark.sql import functions as F

events = (spark.readStream.format("kafka")
               .option("kafka.bootstrap.servers", "broker:9092")
               .option("subscribe", "clicks")
               .load()
               .select(F.from_json(F.col("value").cast("string"), click_schema).alias("e"))
               .select("e.*"))

counts = (events.withWatermark("event_time", "10 minutes")
                .groupBy(F.window("event_time", "5 minutes"), "page_id")
                .count())

# append: each window is written once, after the watermark passes it
(counts.writeStream.outputMode("append")
       .option("checkpointLocation", "s3://lake/_chk/click_counts")
       .format("delta")
       .start("s3://lake/tables/click_counts"))

• Not every sink takes every output mode — the Delta sink takes append and complete; for update semantics use foreachBatch with a MERGE.
• Changing a query’s stateful logic can make an existing checkpoint unusable.

How Spark’s watermark works is §12-2 to §12-6. The engine-neutral concepts (window types, late-data strategies, state backends) and the comparison with Flink and Kafka Streams are in ./stream_pipeline.md.

12-2) Watermarks: what one is and how Spark computes it

A watermark is Spark’s running estimate of how far event time has progressed: the latest event time it has seen, minus the lateness you said you would tolerate. It decides two things — which late rows can still update a result, and which state is old enough to delete.

Every record carries two times. Event time is when the thing happened (a column in the record); processing time is when Spark read it. They drift apart whenever a Kafka consumer lags, a network retries, or a phone goes offline:

Counting by processing time would put C in the wrong window. Counting by event time is right, but then Spark cannot know when a window is complete — C proves an old window can still change. withWatermark is the answer to “how long do we wait”:

# python — tolerate events up to 10 minutes behind the latest one seen
counts = (orders.withWatermark("event_time", "10 minutes")
                .groupBy(F.window("event_time", "10 minutes"))
                .count())

The rule, in Spark 3.5’s micro-batch engine:

• watermark = max(event time seen so far) − delay, where the maximum is taken across every partition of the input.
• It is computed at the end of a micro-batch and used by the next one. Since Spark 3.4 a stateful operator holds two values per batch: it filters late input against the previous batch’s watermark and evicts state against the current one. So in the batch where the watermark first moves, a row whose state lies between the previous value and the new one is still let in; a row behind the previous value is not.
• It only moves forward. An old event never pulls it back.
• It moves only when a batch sees a newer event time. If the whole stream goes quiet, the watermark stops, and the last windows are not finalised until data arrives again.

withWatermark has to name the event-time column the stateful operator uses (the column window() is built on, or a grouping or join column; for dropDuplicates, a column in the dedup key — see §12-4), and it must come before that operator. On a batch DataFrame it is a no-op.

12-3) Late data: what is kept, what is dropped

Spark guarantees that a row less late than the delay — event time > the max seen minus the delay — is never dropped because of the watermark. A row at or beyond that may be dropped or may still be counted; the watermark is a threshold for state, not a strict cut-off for input.

Take 10-minute tumbling windows in append or update mode (in complete mode nothing is evicted, so nothing is dropped either), a 10-minute delay, and a watermark that has been at 10:15 since the previous batch, so the late-row filter and eviction both use 10:15. Three late rows arrive:

B is the row that catches people out. “Older than the watermark means dropped” is not what happens in a window aggregation: what decides it is whether the state the row would update is still there. Spark 3.5 filters a row at a stateful operator when that state is already behind the watermark (for a window, when the window’s end is), and counts it in the progress metric numRowsDroppedByWatermark. The filter’s lag lets C in on one condition: in the batch where the watermark first reaches 10:15, the filter still uses the previous value, so C gets in — and is emitted with its window as that batch evicts it — only if that previous value was still before C’s window end, 10:10 (say 10:05). If it was already 10:12, the filter rejects C in that batch too. The documented contract is only the guarantee above, so do not build on the exact edge.

The dropped rows are not kept anywhere, and a foreachBatch sink after the aggregation sees only aggregated rows, never the ones the operator discarded. If they matter, keep the raw input from before the stateful operator — land the stream in a raw table as well — and reconcile the windows from it in a later batch job. Comparing each row’s event_time with the watermark is not a substitute: it would call B late. Classifying rows inside the stream means reproducing the operator’s own rule, the window’s end against the previous batch’s watermark.

12-4) What a watermark cleans up, and the output modes

A stateful query keeps every open window, unmatched join row or seen dedup key in the state store. Without a watermark nothing ever becomes old enough to delete, so state grows for the life of the query. With one, Spark evicts state once the watermark passes it.

no watermark — every window ever opened stays in the state store
  [10:00–10:10] [10:10–10:20] [10:20–10:30] [10:30–10:40] [10:40–10:50] …

watermark at 10:15 — a window whose end is at or behind 10:15 is final; emit it, then evict it
  [10:00–10:10]  →  emitted (append) and removed
  [10:10–10:20] [10:20–10:30] [10:30–10:40] …  kept: a late row can still update them

Unbounded state shows up as a checkpoint that keeps growing, slower state reads and writes every batch, batch duration creeping up over days, and finally an executor OOM — the reason the RocksDB state store exists (faq_spark_hadoop.md §11-4). A watermark bounds the state of one stateful operator; it does not cap executor memory, and it cannot help an operator that has no event-time column to judge by. dropDuplicates is the usual case: its state is bounded only when the event-time column is part of the key, which dedupes only retries that carry the same timestamp. To dedupe on an ID alone with bounded state, use Spark 3.5’s dropDuplicatesWithinWatermark, which remembers an ID only for the delay: a duplicate whose event time is later than the first copy’s by more than the delay is emitted again, so the delay has to cover the largest gap between duplicates.

The output mode decides when a window’s row reaches the sink, and whether the watermark can delete anything:

An aggregation in append mode needs a watermark — Spark rejects the query without one, because it could never prove a window is final. In update mode a late row inside the delay changes 10:00–10:10 = 100 to 101 and the sink receives the new row; in append mode the sink would have received 101 once, but later.

12-5) Stream–stream joins and multiple watermarks

A stream–stream join keeps both sides in state, waiting for a match. Spark can only delete a buffered row once it knows no future row on the other side can join it — which takes a watermark on each side and an event-time range in the join condition.

# python — a payment must arrive within an hour of its order
orders   = orders_stream.withWatermark("order_time", "10 minutes")
payments = payments_stream.withWatermark("payment_time", "10 minutes")

paid = orders.join(
    payments,
    F.expr("""
        order_id = payment_order_id
        AND payment_time >= order_time
        AND payment_time <= order_time + INTERVAL 1 HOUR
    """),
    "inner",
)

• Inner join — the watermarks and the time range are optional, but without them an order that never gets a payment stays in state forever.
• Outer join — a left outer join needs a watermark on the right side and the time range (the other way round for a right outer join). The right watermark is what lets Spark evict buffered left rows; a watermark on the left is optional, and needed only to evict buffered right rows — without it, right-side state grows. A left row with no match is only emitted, with NULLs, once the watermark proves no match can come, so it arrives late by the time range plus the delay.
• A watermark on each side with no time range in the condition does not bound the state: Spark still cannot tell how long an order may wait for its payment.

When a query reads more than one stream (a join or a union), each input has its own watermark, and the query uses one global watermark chosen by spark.sql.streaming.multipleWatermarkPolicy:

Keep min unless losing the slow stream’s late data is acceptable. The policy is recorded in the checkpoint, so a restarted query keeps the one it started with.

12-6) Watermarks in production: choosing the delay and watching it

Choose the delay from the measured lateness distribution and what the business can lose, not from a round number; then watch the watermark and the state size every batch, because a stuck watermark and growing state are the same incident seen from two ends.

A design prompt, and the reasoning an interviewer wants: success rate per payment method every 5 minutes; Kafka adds ~30 s; 99% of events arrive within 3 minutes; some mobile clients sync up to 15 minutes late; state must stay bounded.

# python
success_rate = (payments.withWatermark("event_time", "15 minutes")
                        .groupBy(F.window("event_time", "5 minutes"), "payment_method")
                        .agg(F.count("*").alias("total"),
                             F.sum(F.when(F.col("status") == "ok", 1).otherwise(0)).alias("ok"))
                        .withColumn("success_rate", F.col("ok") / F.col("total")))

Then the questions that settle it: is 15 minutes an observed p99.9 or a guess? Do some devices sync hours later? Does the business accept an approximate number now and a corrected one later — update mode, or a batch job that recomputes yesterday from the lake? Where do rows behind the watermark go?

Watch it — every StreamingQueryProgress carries the watermark and the state operators:

# python
p = query.lastProgress
p["eventTime"]["watermark"]                    # e.g. "2026-11-08T10:15:00.000Z"
p["stateOperators"][0]["numRowsTotal"]         # rows held in state
p["stateOperators"][0]["numRowsRemoved"]       # evicted this batch
p["stateOperators"][0]["numRowsDroppedByWatermark"]  # late input discarded
p["stateOperators"][0]["memoryUsedBytes"]
p["durationMs"]["triggerExecution"]            # batch duration

One Spark-specific trap. Spark’s watermark comes from the maximum across all partitions, so an idle Kafka partition does not hold it back. Flink takes the minimum across source instances, where an idle partition does freeze the watermark (stream_pipeline.md §3-3). The Spark failure runs the other way: one partition far ahead — a fast producer, a backfill, a skewed clock — advances the watermark for everyone, and the slower partitions’ rows turn up late and get dropped.

13) Tuning Checklist ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

13-1) In the order you would check

Find the slow stage first, then ask whether it reads too much, shuffles too much, or has one task doing all the work — tune configs last.

1. Read less — partition pruning on the filter column, column pruning (no SELECT *), pushdown not blocked by a UDF, columnar formats.
2. Shuffle less — broadcast small dimensions, drop needless repartition / distinct, aggregate before joining.
3. Fix skew — AQE skew join, isolate hot keys, salt.
4. Right-size partitions — ~100–200 MB each, enough tasks for every core; AQE on.
5. Avoid Python in the hot path — built-ins over UDFs, pandas_udf over udf.
6. Cache what is reused, unpersist after.
7. Fix file layout — compact small files, sensible partitionBy, bucketing / clustering for repeated joins.
8. Then size resources — executor cores / memory / overhead, dynamic allocation.

13-2) The configs worth knowing by name

14) Debugging with the Spark UI ★★★★☆Priority 4 of 5 — High value — a gap here costs you rounds

14-1) Where to look

Jobs tab → the slow job; Stages tab → the slow stage and its task-time distribution; SQL tab → the physical plan with real row counts; Executors tab → GC time, spills and failures.

The Spark History Server shows the same UI for finished applications.

14-2) Reading the symptoms

15) Common Failures and Fixes ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

15-1) Out of memory

15-2) Other errors worth recognising

• Task not serializable — a closure captured a non-serialisable object; build it inside mapPartitions / foreachPartition, or mark it as a broadcast.
• AnalysisException: cannot resolve column — typo, a column dropped upstream, or an ambiguous column after a self-join (alias both sides).
• Job “hangs” at 199/200 tasks — the last task is the skewed one.
• Timeouts on broadcast (spark.sql.broadcastTimeout, 300s) — the “small” side is not small, or is slow to compute; materialise it or stop broadcasting it.
• Different results on re-run — non-deterministic code (rand() without a seed, monotonically_increasing_id, first() without an order) recomputed after a failure.
• Out of disk — shuffle and spill files fill local disks; give executors more local storage or shuffle less.

15-3) Exactly-once is your job in batch too

A Spark batch job is retried at task, stage and application level, so its writes must be idempotent — re-running must replace, not add.

• Writes: dynamic partition overwrite or MERGE keyed on a natural key, not append.
• Side effects in tasks (HTTP calls, emails, writing to an external DB in foreach) can run more than once on retry or speculation — make them idempotent or move them out of the job.
• Output committers: on object stores, use the S3A committers or a table format rather than rename-based commits, which are not atomic on S3.

16) Frequently Asked Questions ★★★★★Priority 5 of 5 — Must know — expect it in almost every loop

16-1) Quick-fire answers

• What triggers a stage boundary? A wide dependency — a shuffle.
• How many tasks does a stage have? One per partition of that stage.
• cache() vs persist()? cache() is persist() at the default level (MEMORY_AND_DISK for DataFrames, MEMORY_ONLY for RDDs).
• repartition vs coalesce? Full shuffle, any count vs merge without a shuffle, only fewer.
• groupByKey vs reduceByKey? reduceByKey combines on the map side first; prefer it.
• Why is my job slow after one stage finished fast? Usually skew or a shuffle that is too large for its partition count.
• What does AQE do? Re-plans at shuffle boundaries: coalesces partitions, switches to broadcast joins, splits skewed partitions.
• What is predicate pushdown? Moving a filter into the scan, so the source skips data.
• Broadcast join limit? spark.sql.autoBroadcastJoinThreshold, 10 MB by default; the small side must fit in driver and executor memory.
• How does Spark recover a lost partition? Recomputes it from lineage.
• Narrow vs wide? Narrow: one parent partition per child partition. Wide: many, so a shuffle.
• Serialisation vs deserialisation? Object → bytes to ship or store; bytes → object to compute on. Spark does both on every shuffle, broadcast, serialised cache and task launch.
• Does Kryo speed up a DataFrame shuffle? No. SQL exchanges serialise UnsafeRows with their own serialiser; Kryo helps RDD records and broadcasts.
• What causes Task not serializable? A closure captured something that cannot be serialised. Create it on the executor, inside foreachPartition.
• Why avoid Python UDFs? Row-by-row serialisation to a Python process, and the optimiser cannot see inside them.
• client vs cluster deploy mode? Where the driver runs: your machine vs inside the cluster.
• What is a DAG in Spark? The graph of RDD / DataFrame dependencies the scheduler cuts into stages.
• Accumulators — any caveat? Updates in transformations can be counted more than once on retry.
• Static vs dynamic partition overwrite? Static replaces the whole table; dynamic only the partitions present in the written data.
• What is a shuffle spill? Execution memory filled up, so the operator wrote sorted runs to local disk and merged them later.
• Window function with no partitionBy? All rows move to one partition — one task does everything.
• How do you pick the number of partitions? Total data ÷ ~128 MB, at least 2–3× the cores, then let AQE coalesce.
• What is a watermark? Max event time seen minus the allowed delay. Windows and join rows behind it are final, and their state is deleted.
• Does a 10-minute watermark drop everything older than 10 minutes? No. Data within the delay is never dropped; data later than it may be — a row whose window is still open is counted.
• Watermark in complete mode? It cleans nothing: the whole result is kept.
• Two input streams, which watermark? The minimum of the two by default (spark.sql.streaming.multipleWatermarkPolicy).
• Spark on Kubernetes vs YARN? Same engine; K8s runs each executor as a pod, has no external shuffle service by default, so dynamic allocation uses shuffle tracking.

16-2) Design question: “Process 10 TB of daily clickstream”

State the shape of the answer first: read only the day’s partition, clean and dedupe, join small dimensions by broadcast, aggregate, and write partitioned, compacted output with a re-runnable overwrite.

1. Input — clicks land in s3://lake/raw/clicks/dt=YYYY-MM-DD/ as Parquet (or convert JSON to Parquet in a first step with an explicit schema).
2. Read — filter on dt = run_date, select only the needed columns.
3. Clean — parse, drop malformed rows to a quarantine path, dedupe on event_id (dropDuplicates(["event_id"]) within the day).
4. Enrich — broadcast-join small dimensions (pages, campaigns); sort-merge join users if large.
5. Aggregate — sessions and daily metrics with groupBy and window functions partitioned by user_id.
6. Write — Delta / Iceberg (or Parquet with dynamic partition overwrite), partitioned by dt, repartition by dt (plus a clustering key) to control file sizes.
7. Operate — AQE on, shuffle partitions sized for ~10 TB (≈ several thousand), data-quality checks before publishing, orchestrated with a logical date for backfills.