200 Apache Spark interview questions, with answers

A distributed engine for large-scale data processing. You express a computation over a dataset, and Spark splits both the data and the work across a cluster, coordinating it from a single driver process. It provides APIs in Scala, Java, Python and R, plus libraries for SQL, streaming, machine learning and graphs.

Because the data no longer fits the memory, CPU or disk of one machine, or because the work finishes sooner spread across many. Spark divides a dataset into partitions and processes them in parallel. The trade is coordination cost, which is why a small dataset often runs faster on one machine than on a cluster.

MapReduce writes intermediate results to disk between every map and reduce. Spark keeps them in memory where it can, and expresses a whole chain of operations as one plan before running any of it. That matters most for iterative work, where MapReduce re-reads the same data from disk on every pass.

The honest framing for an interview: Spark’s advantage is in-memory intermediate data plus whole-plan optimisation, not a single fixed speed multiplier. Quote a measured number only if you measured it on your own data.

Transformations do not compute anything when you call them. They record what to do and return a new dataset. Nothing runs until an action asks for a result. The RDD programming guide states it directly: “All transformations in Spark are lazy, in that they do not compute their results right away.”

This is what makes optimisation possible: Catalyst sees the whole chain before any of it has run.

A transformation creates a new dataset from an existing one and runs nothing. An action returns a value to the driver or writes to storage, and submits a job.

Count the jobs. Chaining transformations submits none; the action submits one.

rdd = spark.sparkContext.parallelize(range(4), 2)
doubled = rdd.map(lambda x: x * 2).filter(lambda x: x > 2)
print(len(spark.sparkContext.statusTracker().getJobIdsForGroup() or []))   # 0
doubled.count()
print(len(spark.sparkContext.statusTracker().getJobIdsForGroup() or []))   # 1

The unit of parallelism. A dataset is split into partitions, and a stage runs one task per partition. Almost every Spark performance question reduces to how many partitions you have and how evenly the data is spread across them.

SparkSession is the entry point for the DataFrame and SQL APIs, and it is what you create in modern code. SparkContext is the lower-level entry point for RDDs and cluster coordination, reachable as spark.sparkContext.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("trips").getOrCreate()
sc = spark.sparkContext

One SparkSession wraps one SparkContext.

Scala, Java, Python and R. For DataFrame and SQL work the language barely matters, because the operations compile to the same plan and run in the JVM. It matters for RDDs and Python UDFs, where data must cross into a Python process and back, which blocks whole-stage codegen and predicate pushdown across that point.

A job has one or more stages, a stage has one task per partition.

A directed acyclic graph of dependencies between datasets. Directed because each edge points from input to output, acyclic because nothing depends on itself. DAGScheduler walks it to cut stages and to work out what to recompute if something is lost.

The process that holds the SparkSession, builds and optimises the plan, cuts it into stages, schedules every task, and tracks the results. It is the coordinator and it does not process your data. If the driver dies, the application dies, because the plan and the scheduler state die with it.

A JVM launched for one application that runs tasks and stores cached and shuffle blocks. The source describes it in a line: a “Spark executor, backed by a threadpool to run tasks”. It never makes a scheduling decision and never sees the query plan.

A separate service that grants containers. At Spark 4.x the options are Standalone, YARN and Kubernetes. Mesos support was removed in the 4.x line: resource-managers/ in the source tree holds only kubernetes and yarn.

The key division is that the cluster manager grants resources but does not schedule your tasks. The driver does that.

In the general sense yes, one coordinator and many workers, but Spark’s own vocabulary matters and the word “master” is overloaded.

Spark also moved away from the older terminology: at v4.2.0 there is sbin/start-workers.sh and conf/workers.template, while start-slaves.sh and slaves.template are gone.

The script that launches an application against any supported cluster manager through one interface.

spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 4g \
  --executor-cores 4 \
  examples/jars/spark-examples.jar 1000

Every Spark-facing option must come before the application jar. Anything after it is passed to your application instead.

Where the driver runs. In client mode it runs in the submitting process; in cluster mode the cluster manager launches it inside the cluster.

One consequence catches people: in client mode the driver JVM already exists by the time your code runs, so spark.driver.extraJavaOptions set in SparkConf arrives too late. Use --driver-java-options.

spark.default.parallelism governs RDD shuffle operations. The documented default depends on the cluster manager: for distributed shuffles it is the largest number of partitions in the parent RDD, and for operations like parallelize with no parent it follows the cluster manager, which on local mode is the number of cores. For DataFrames, the relevant knob is spark.sql.shuffle.partitions, which defaults to 200.

The number of partitions produced by a DataFrame shuffle. It defaults to 200 regardless of your data size or cluster, which is why it is wrong so often: 200 tasks over a few megabytes is scheduling overhead, and 200 tasks over a terabyte is spill. Adaptive query execution coalesces it down at runtime, which is why leaving the default is more tolerable in modern Spark than it used to be.

Spark 3.3.0. The file became log4j2.properties and the flag -Dlog4j.configurationFile=. At v4.2.0 the only template shipped is conf/log4j2.properties.template; the Log4j 1 template is gone. Guides that tell you to edit log4j.properties with log4j.rootLogger= predate that change and the file will be ignored.

A Resilient Distributed Dataset: an immutable, partitioned collection that can be operated on in parallel, with enough information recorded to rebuild any lost partition. Resilient because of lineage, distributed because of partitions.

From RDD.scala, every RDD is defined by:

1. A list of partitions.
2. A function for computing each partition.
3. A list of dependencies on other RDDs.
4. Optionally, a partitioner for key-value RDDs.
5. Optionally, preferred locations for each partition.

Properties 4 and 5 are the ones that explain shuffle avoidance and data locality.

Through lineage. Spark records how each partition was derived rather than only the partition itself, so a lost partition is recomputed from its parents on a surviving executor.

lineage = (spark.sparkContext.parallelize(range(10), 2)
    .map(lambda x: x + 1)
    .filter(lambda x: x % 2 == 0))
print(lineage.toDebugString().decode())

Recovery is per partition, not per job.

In a narrow dependency each parent partition feeds at most one child partition, so the work pipelines inside one task. In a wide dependency a child partition draws from many parents, which requires a shuffle and therefore a stage boundary.

At every ShuffleDependency in the lineage, and nowhere else. DAGScheduler walks the graph backwards from the action and starts a new stage each time it crosses one. Everything narrow pipelines into the stage it is already in, which is why a long chain of map and filter costs no extra stages.

reduceByKey combines values on the map side before the shuffle, so far less data crosses the network. groupByKey shuffles every value and combines afterwards.

pairs.reduceByKey(lambda a, b: a + b)      # prefer this
pairs.groupByKey().mapValues(sum)          # shuffles everything

The subtlety worth knowing: a map-side combine disqualifies Spark’s two faster shuffle writers, so reduceByKey shuffles less data through a more expensive writer. It is still usually the right choice, because less network traffic dominates.

map applies a function per element; mapPartitions applies it once per partition, receiving an iterator. Use mapPartitions when there is per-partition setup worth amortising, such as opening a database connection, and be careful not to materialise the whole partition in memory.

map returns one output per input. flatMap returns zero or more per input and flattens the result, which is how you both expand and filter in one pass.

repartition shuffles to reach any target number of partitions and balances them. coalesce merges partitions without a full shuffle, so it can only reduce the count and may leave them uneven.

Use coalesce to reduce partitions before a write; use repartition when you need balance or more partitions. One trap: coalesce(1) before a wide transformation can shrink the parallelism of the whole upstream stage, because it removes the shuffle boundary that would have isolated it.

cache() is persist() with the default storage level. persist(level) lets you choose. The default differs by API, which surprises people:

So an RDD cache that does not fit simply drops partitions and recomputes them, while a DataFrame cache spills to disk.

MEMORY_ONLY, MEMORY_ONLY_SER, MEMORY_AND_DISK, MEMORY_AND_DISK_SER, DISK_ONLY, their _2 replicated variants, and OFF_HEAP. The _SER variants store a serialized form, which is more compact and more CPU to read. The _2 variants replicate to a second node so a lost executor does not force recomputation.

A read-only value cached once per executor rather than shipped with every task. The guide puts it as keeping “a read-only variable cached on each machine rather than shipping a copy of it with tasks”.

cities = sc.broadcast({"sf": "San Francisco", "nyc": "New York"})
resolved = sc.parallelize(["sf", "nyc"]).map(lambda c: cities.value[c])

Use them for lookup tables. Without one, a large captured object is serialized per task.

Write-only counters that tasks add to and the driver reads. The catch is the guarantee: exactly-once applies only to updates made inside actions. The guide says that in transformations “each task’s update may be applied more than once if tasks or job stages are re-executed”.

acc = sc.accumulator(0)
counted = sc.parallelize(range(100), 4).map(lambda x: (acc.add(1), x)[1])
counted.count(); print(acc.value)   # 100
counted.count(); print(acc.value)   # 200, same 100 rows

The uncached dataset is recomputed, so the accumulator increments again. Add .cache() and it stays at 100. Use accumulators for diagnostics, never for a number the output depends on.

The function mapping a key to a partition, HashPartitioner or RangePartitioner. It helps when two datasets share a partitioner on the join key, because the join then needs no shuffle. Persist a partitioned RDD if you will reuse it, or the partitioning is recomputed.

The lineage of an RDD as an indented tree, with each level showing the RDD type and the number of partitions. Indentation changes mark shuffle boundaries, so it is the quickest way to see how many stages a computation will take.

When you need control the structured APIs do not expose: custom partitioning, per-partition resource handling, or operations on unstructured data with no schema. For anything with a schema, the DataFrame API goes through Catalyst and will usually beat hand-written RDD code.

collect() brings every row to the driver, bounded only by spark.driver.maxResultSize, which defaults to 1g. take(n) returns n rows and starts by scanning one partition, adding more only if needed. Prefer take for inspection; collect on a large result is the classic way to kill a driver.

Both aggregate to a single value. fold takes a zero value and so is defined on an empty dataset, where reduce throws. The zero value must be a true identity, because it is applied per partition as well as across them.

Because only they can be partitioned by key, which is what enables shuffle-free joins, reduceByKey-style map-side combines, and lookup. In Scala they are available through implicit conversion on RDD[(K, V)]; in Python any RDD of 2-tuples qualifies.

It is retried, up to spark.task.maxFailures, which defaults to 4, before the stage fails. If instead a completed stage’s shuffle output was lost with its executor, DAGScheduler resubmits that whole stage, because the next stage has nothing to fetch. That is a stage rerun rather than a task retry, and it is why a job can appear to go backwards in the UI.

A distributed collection of rows organised into named columns, with a schema. Operations on it go through Catalyst, so Spark can reorder filters, prune columns and choose join strategies. It is the API to default to.

A Dataset is a typed collection that still runs through Spark SQL’s optimised engine. In Scala and Java it gives compile-time type safety and encoders. In Python there is no typed Dataset API, because the language has no compile-time types to check, so DataFrame is what you get.

Spark SQL’s query optimiser. It takes a parsed query through resolution, rule-based logical optimisation, physical planning with cost-based choices, and then code generation. The important idea is that it optimises a whole plan, which is only possible because transformations are lazy.

1. Parse into an unresolved logical plan.
2. Analyse, resolving tables, columns, functions and types against the catalog.
3. Optimise the logical plan with rules: predicate pushdown, column pruning, constant folding, filter reordering.
4. Plan physically, choosing join strategies and producing one or more physical plans.
5. Generate code, compiling operators into JVM bytecode via whole-stage codegen.

Moving a filter as close to the data source as possible, so fewer rows are read. With Parquet, Spark can push it into the file format, which skips row groups by their footer statistics. You can see it in the plan as PushedFilters.

FileScan parquet [fare_amount,city_id]
  PushedFilters: [IsNotNull(fare_amount), GreaterThan(fare_amount,50.0)]

Reading only the columns the query needs. It is a large win on columnar formats such as Parquet and ORC, where unread columns are never touched on disk. Selecting explicitly rather than using select("*") is what lets it happen.

Catalyst fuses the operators in a stage into a single generated Java method, so intermediate rows never become objects and the loop is tight. In explain output, fused operators are marked with a * and a codegen stage id. Python UDFs break the fusion, which is a large part of their cost.

df.explain("formatted")     # the readable one
df.explain("extended")      # parsed, analysed, optimised, physical
df.explain("cost")          # with statistics, if available

Read the physical plan bottom up: scans at the bottom, then filters and projections, then exchanges, then the final operator. Each Exchange is a shuffle and therefore a stage boundary.

select projects a set of columns; withColumn adds or replaces one, keeping the rest. Chaining many withColumn calls builds a deep plan and is slower to analyse than one select with all the expressions, which matters when you are adding dozens of columns.

createOrReplaceTempView is scoped to the SparkSession that created it. createGlobalTempView is visible to every session in the application, under the global_temp database, and lives until the application ends.

A user-defined function. Avoid it when a built-in exists, because a UDF is a black box: Catalyst cannot see inside it, so it cannot push filters through it or fuse it into generated code. In PySpark a plain Python UDF also serialises rows out to a Python process and back.

A vectorised UDF that operates on Arrow batches rather than one row at a time, so the interpreter overhead is amortised across the batch. Use it when you genuinely need Python logic on columns. It still breaks codegen, but the per-row cost is far lower than a plain Python UDF.

Use isNull and isNotNull rather than equality, because col = NULL is never true in SQL. na.fill, na.drop and coalesce cover most cases. Be aware that from Spark 4.0 spark.sql.ansi.enabled is on by default, which changes how invalid casts and overflows behave: they raise instead of returning null.

The migration guide is explicit: “Since Spark 4.0, spark.sql.ansi.enabled is on by default.” Invalid casts, numeric overflows and division by zero now raise errors instead of yielding null. Set it to false to restore the old behaviour, but treat a failure on upgrade as a finding: it is usually pointing at data that was silently becoming null before.

union matches columns by position; unionByName matches by name. Positional matching on two DataFrames whose columns are in different orders silently produces wrong data, which makes unionByName the safer default. It also takes allowMissingColumns.

(df.write
   .partitionBy("city_id")
   .mode("overwrite")
   .parquet("s3a://lakehouse-prod/warehouse/trips"))

The trap is file count. With 200 shuffle partitions and 50 cities you can get 10,000 small files. Repartition by the same column first so each partition value is written by few tasks.

partitionBy writes a directory per value, which prunes at scan time. bucketBy hashes rows into a fixed number of files per partition and records it in the metastore, which lets a join on the bucket key skip the shuffle. Bucketing requires a metastore table, so it only applies to saveAsTable.

append, overwrite, ignore and error (the default, also spelled errorifexists). With partitioned output, spark.sql.sources.partitionOverwriteMode set to dynamic makes overwrite replace only the partitions the data touches rather than the whole table.

It is columnar, so column pruning skips unread columns on disk; it carries per-row-group statistics, so pushed-down predicates skip data; and it compresses well because a column holds like-typed values. spark.sql.parquet.filterPushdown defaults to true.

Give Spark a numeric, date or timestamp column to split on, plus bounds.

trips = (spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://db.internal:5432/rides")
    .option("dbtable", "public.trips")
    .option("partitionColumn", "trip_seq")
    .option("lowerBound", "1")
    .option("upperBound", "40000000")
    .option("numPartitions", "16")
    .load())

Without partitionColumn the whole table is read by a single task. The column must be numeric, date or timestamp: a string key will not do.

Caching stores the computed result; a view stores the query. CACHE TABLE in SQL caches eagerly by default, whereas df.cache() is lazy and populates on the next action.

1. spark-submit hands the application to the cluster manager.
2. The driver starts and creates the SparkSession.
3. The driver asks the cluster manager for executors.
4. Executors start and register back with the driver.
5. Your code builds a plan; nothing runs yet.
6. An action submits a job.
7. Catalyst optimises; DAGScheduler cuts stages at shuffle boundaries.
8. TaskSchedulerImpl places tasks on executors by locality.
9. Executors run tasks and write shuffle output for the next stage.
10. Results return to the driver, or executors write them to storage.

Creates the SparkSession; requests executors through SchedulerBackend; turns code into a logical then physical plan; cuts stages in DAGScheduler; places tasks via TaskSchedulerImpl and TaskSetManager; tracks attempts and retries; keeps the block index in BlockManagerMaster; receives results; and serves the UI on port 4040.

Registers with the driver by sending RegisterExecutor and waiting for RegisteredExecutor; accepts LaunchTask messages and runs each in a pool thread; reads input and writes shuffle output; reports StatusUpdate per task plus a heartbeat every spark.executor.heartbeatInterval, default 10s; and stores and serves blocks through its BlockManager.

Until that registration round trip completes, a granted container is doing nothing.

Decides who gets containers, and nothing else. Standalone uses a Master tracking Worker daemons; YARN uses the ResourceManager, asked by Spark’s ApplicationMaster through YarnAllocator; on Kubernetes the driver’s own ExecutorPodsAllocator creates pods, so there is no Spark cluster-manager daemon at all.

Turning a job into a DAG of stages by cutting at every ShuffleDependency, submitting each stage as a TaskSet, and resubmitting stages whose shuffle output has been lost. It is the stage-level scheduler.

Placing individual tasks from a TaskSet onto executors, tracking attempts, handling retries and speculation. It is the task-level scheduler and it works with TaskSetManager per stage.

The one component that talks to the cluster manager, with an implementation per manager. It requests and releases executors and relays their registration to the scheduler.

Every stage but the last is a ShuffleMapStage, whose job is to produce shuffle output for the next stage. The last is a ResultStage, which computes the answer the action asked for. That is why the UI shows a final stage with a different shape.

PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY. The scheduler prefers the best available and waits briefly for it, governed by spark.locality.wait. If your data is remote object storage, locality mostly does not apply and lowering the wait can help.

Not from the number of files, which is the assumption that costs most people time. FilePartition.maxSplitBytes decides the split size:

Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))

spark.sql.files.maxPartitionBytes defaults to 128MB and spark.sql.files.openCostInBytes to 4MB, charged per file so small files are packed together. The consequence: the same data written as 1, 8 or 16 Parquet files can read back as 1, 2 and 2 partitions. Read the number with getNumPartitions() rather than predicting it.

One per partition of that stage. For the first stage that is the input partition count; after a shuffle it is spark.sql.shuffle.partitions, or whatever adaptive execution coalesced it to. It is not related to the number of executors.

Its running tasks are retried elsewhere and the partitions it held are rebuilt from lineage. Cached blocks are recomputed unless the storage level replicated them. If a completed stage’s shuffle output died with it, that stage is resubmitted.

Because the driver’s state, the plan, the scheduler and the block index, exists nowhere else. Executors hold only data that can be recomputed or refetched, so they are disposable by design.

A daemon on each node that serves shuffle blocks independently of the executors that wrote them, so an executor can exit without destroying its shuffle output. spark.shuffle.service.enabled defaults to false. It matters for dynamic allocation, which otherwise cannot release an executor still holding shuffle data.

Adding and removing executors according to pending work. spark.dynamicAllocation.enabled defaults to false, so absent a decision your application holds its executors from first request to exit. Without an external shuffle service it needs spark.dynamicAllocation.shuffleTracking.enabled, which defaults to true, to avoid discarding executors that still hold shuffle blocks.

Re-running tasks that are running much slower than their peers and taking whichever finishes first. spark.speculation defaults to false. It helps with a slow node; it hurts when the task is slow because of skew, since the duplicate is equally slow and wastes a slot.

spark.scheduler.mode defaults to FIFO, so jobs run in submission order and a large job can starve later small ones. FAIR shares slots between jobs and is worth setting when one application serves many concurrent queries, such as a notebook or a JDBC server.

spark.driver.extraJavaOptions and spark.executor.extraJavaOptions. In client mode the driver JVM is already running by the time SparkConf executes, so pass driver options with --driver-java-options or a properties file instead. Both configs are the route for logging configs, heap dump flags, stack size and proxy settings.

A cap on the total size of results returned to the driver, default 1g. It exists because collect() on a large dataset would otherwise exhaust the driver heap. Raising it treats the symptom; the fix is usually not to collect.

spark.ui.port defaults to 4040. When the application ends the UI goes with it, which is why you enable event logging and run a history server to replay it afterwards.

Broadcast variables go outward, cached once per executor rather than per task. Accumulators come back, aggregated on the driver. They exist because a closure’s captured variables are serialized to the executor and updates to them never return.

A client-server split where the client sends an unresolved logical plan over gRPC and the server runs it. The practical consequence is that the client no longer needs to be a JVM colocated with the driver, so a thin client can drive a remote cluster and a crash in user code cannot take out the driver.

A redistribution of data across partitions so that rows which must be processed together end up together. The map side writes partitioned, sorted output to local disk; the reduce side fetches its slice from every map output over the network. It is the most expensive thing Spark does.

Any wide dependency: groupBy, join without a broadcast, distinct, repartition, orderBy, reduceByKey, window functions without an existing matching partitioning. In the physical plan every shuffle appears as an Exchange.

Because it involves serialization, disk writes on the map side, network transfer, and often sorting or merging on the reduce side. It also creates a stage boundary, so the downstream stage cannot start until the upstream one has finished writing.

SortShuffleManager picks between three, cheapest first:

A map-side combine, which is exactly what reduceByKey does, disqualifies the two fast paths immediately.

An uneven distribution of rows across partitions, usually because one key dominates. The symptom is a stage where most tasks finish quickly and one or two run far longer, since a task cannot be split.

In the Spark UI’s stage page, compare the max task duration and shuffle read size against the median. A large gap is skew. In the data, count rows per key and look at the top of the distribution.

In order of preference:

1. Enable adaptive skew join handling, which splits skewed partitions automatically.
2. Broadcast the small side if the join allows it, removing the shuffle.
3. Salt the hot key: add a random suffix to spread it, join on the salted key, then aggregate away the salt.
4. Process the hot keys separately and union the results.

Salting is the general answer when the hot key is genuinely large.

from pyspark.sql import functions as F

N = 16
facts = trips.withColumn("salt", (F.rand() * N).cast("int"))
dims  = (cities
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)]))))

joined = facts.join(dims, ["city_id", "salt"]).drop("salt")

The fact side gets a random salt; the dimension side is replicated once per salt value so every match still exists. The cost is N times the dimension rows, which is why N stays small.

Both conditions must hold: a partition is treated as skewed only when it is larger than 5 times the median and larger than 256 MB. On a job whose partitions are all under 256 MB, skew handling never fires no matter how uneven they are.

Re-planning parts of a query at runtime using statistics from completed stages. It coalesces shuffle partitions, converts sort-merge joins to broadcast joins when a side turns out small, and splits skewed partitions. It is on by default since Spark 3.2.

After a shuffle, it merges small output partitions into fewer, larger tasks, targeting spark.sql.adaptive.advisoryPartitionSizeInBytes, default 64MB, with a floor of spark.sql.adaptive.coalescePartitions.minPartitionSize, default 1MB. This is what makes the static 200 default tolerable.

Aim for partitions in the low hundreds of megabytes, and for a total that is a small multiple of your total cores so every core gets several tasks. With AQE enabled, set the advisory size rather than the partition count, and let coalescing do the arithmetic.

Many tiny output files make later reads slow, because planning cost and per-file open cost dominate. It comes from writing with high parallelism or over-partitioning. Fix it by coalesce or repartition before the write, by partitioning output on a low-cardinality column, or by setting spark.sql.files.maxRecordsPerFile, which defaults to 0 meaning unlimited.

repartition(n) distributes rows round-robin into n partitions. repartition(col) hash-partitions by that column, so all rows with the same value land together. The second is what you want before writing partitioned output or before a join on that key.

Yes. A global sort uses a RangePartitioner, which first samples the data to find range boundaries and then shuffles rows into the right range. That sampling is an extra job, which is why a global sort is more expensive than it looks. sortWithinPartitions avoids it when you only need local ordering.

Three ways: broadcast the small side; join on a column both tables are already bucketed by, with matching bucket counts; or, with RDDs, join two datasets that already share a partitioner.

The reduce side pulling blocks from map outputs. spark.reducer.maxSizeInFlight bounds how much data is in flight per reduce task, and spark.shuffle.io.maxRetries with spark.shuffle.io.retryWait govern retries. FetchFailedException almost always means an executor that held shuffle output died, not a network misconfiguration.

The executor holding the shuffle blocks is gone, was killed by the resource manager for exceeding memory, or is too busy to serve. Look at why that executor died rather than at fetch settings. An external shuffle service makes shuffle output survive the executor.

Data written to disk because the shuffle could not hold it in execution memory. The UI reports spill in memory and on disk. Spill means the partitions are too large for the memory available, so raise parallelism or lower the advisory partition size rather than only raising memory.

Because it removes a shuffle boundary, so the reduced parallelism propagates upstream: the whole computation feeding it may run in one task. If you need exactly one output file, repartition(1) pays for a shuffle but keeps the upstream stage parallel.

UnifiedMemoryManager subtracts a fixed 300 MB of reserved memory, then takes spark.memory.fraction of what remains, default 0.6, as the unified pool for execution and storage. The rest is user memory for your own data structures. Off-heap is separate and off by default: spark.memory.offHeap.enabled defaults to false.

Execution memory is for shuffles, joins, sorts and aggregations. Storage memory is for cached blocks and broadcast data. They draw from one unified pool, and spark.memory.storageFraction, default 0.5, sets the floor below which cached blocks will not be evicted.

The eviction is one-way. Execution can evict cached blocks down to the storage floor; storage can never evict execution. So caching a large DataFrame cannot fail a shuffle, but it can make one spill, which is how adding a cache() makes a job slower.

On-heap memory is managed by the JVM and subject to garbage collection. Off-heap is allocated outside the heap, bounded by spark.memory.offHeap.size, and avoids GC pressure at the cost of explicit management. Tungsten’s binary formats are what make off-heap useful.

Memory requested from the cluster manager beyond the JVM heap, for off-heap allocations, the Python worker, and native libraries. spark.executor.memoryOverhead defaults to a fraction of executor memory with a minimum floor. Containers killed for exceeding memory limits are usually an overhead problem, not a heap problem.

Usually collect() or toPandas() on a large result, a broadcast of something that is not small, or a very large number of tasks whose metadata the driver must track. spark.driver.memory defaults to 1g, which is fine for a coordinator and not for anything pulling data back.

A partition too large to process, a skewed key concentrating rows in one task, a wide aggregation with high cardinality, or a UDF materialising too much. Raising memory is the last resort: raising the partition count is usually the fix, because it makes each task’s share smaller.

When a DataFrame is used more than once and producing it was expensive. Caching something read once makes the job slower, because you pay the write and give up memory that execution wanted.

When the dataset is used once, when it is cheap to recompute, or when memory is already tight. Also avoid caching immediately before a single wide shuffle, where the cache competes with the shuffle for the same pool.

MEMORY_AND_DISK is the default for df.cache(). MEMORY_AND_DISK_SER trades CPU for a smaller footprint. DISK_ONLY is for when recomputation is more expensive than reading from disk. MEMORY_ONLY is the RDD default and will silently drop partitions that do not fit.

df.unpersist(), and spark.catalog.clearCache() for everything. It matters because cached blocks hold storage memory for the life of the application otherwise, squeezing execution memory for every later stage.

No. It marks the dataset to be cached and populates on the next action. CACHE TABLE in SQL is eager by default, which is the exception.

Checkpointing writes the dataset to reliable storage and truncates the lineage, so recovery does not replay the whole chain. Caching keeps lineage intact and stores a copy for reuse. Use checkpointing for very long lineages, such as iterative algorithms, where recomputation would be prohibitive.

localCheckpoint() truncates lineage using executor storage: faster, but lost if the executor is lost.

Because the plan itself grows, analysis and planning take longer, and a single lost partition triggers a long recomputation. A StackOverflowError from a deep plan is the extreme version, and checkpointing is the standard remedy.

spark.serializer defaults to org.apache.spark.serializer.JavaSerializer. Kryo is more compact and faster, and is worth setting for RDD-heavy or shuffle-heavy workloads; register your classes for the full benefit. DataFrame operations mostly use Tungsten’s own encoders, so the effect there is smaller.

The execution backend that stores rows in a compact binary format off-heap, operates on them without deserialising into JVM objects, and works with whole-stage codegen. It is why the DataFrame API beats hand-written RDD code on structured data.

Look at the GC time column in the executor tab of the UI. Consistently high GC time relative to task time points at too many small objects, too much data per executor, or a heap sized so large that full collections are slow. Serialized storage levels and fewer, smaller partitions per executor both help.

Which datasets are cached, the fraction actually held in memory, and how much spilled to disk. A cached dataset showing well below 100% in memory is one that is being partly recomputed or read from disk on every use, which usually means the cache is not paying for itself.

Spark ships the smaller side to every executor so the join needs no shuffle. It chooses this automatically when the estimated size is below spark.sql.autoBroadcastJoinThreshold, default 10485760, which is 10 MB. Set it to -1 to disable.

from pyspark.sql.functions import broadcast
joined = trips.join(broadcast(cities), "city_id")

Or with a hint in SQL: SELECT /*+ BROADCAST(c) */ .... Available hints include BROADCAST, MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL. Forcing a broadcast of something too large moves the failure to the driver, which must collect it first.

How long a broadcast may take to build before the query fails, default 300 seconds. Hitting it usually means the broadcast side is not actually small, so raise the estimate quality or change strategy rather than the timeout.

Because it needs only a sort per partition and a single merge pass, with memory proportional to the merge buffers rather than to a whole hash table. spark.sql.join.preferSortMergeJoin defaults to true.

Both sides are shuffled, then a hash table is built from the smaller side per partition. It avoids the sort, so it wins when one side is small enough to hash per partition but too large to broadcast. Spark chooses it less often because it is more memory-hungry per task.

A bucketed join avoids the shuffle because both tables were written hashed into the same number of buckets on the join key, recorded in the metastore. A broadcast avoids it by copying the small side everywhere. Bucketing works for large-to-large joins; broadcasting does not.

Because Spark’s size estimate is based on statistics that may be missing or stale, especially for a non-catalog source or after a chain of transformations. Run ANALYZE TABLE ... COMPUTE STATISTICS, or add a hint. Adaptive execution helps here, because it can switch to a broadcast at runtime once the real size is known.

Join reordering and strategy selection driven by table and column statistics. spark.sql.cbo.enabled defaults to false, and it also needs statistics to have been computed. Adaptive execution has taken over much of what CBO was meant to do, using real runtime numbers instead of estimates.

After a shuffle stage completes, it knows the actual size of each side, so it can demote a planned sort-merge join to a broadcast hash join. This is the single most useful thing AQE does, because planning-time estimates on derived data are so often wrong.

An AQE optimisation: when a sort-merge join becomes a broadcast join at runtime, the shuffle that was already written can be read locally instead of over the network. spark.sql.adaptive.localShuffleReader.enabled defaults to true.

Compare df.explain() before execution with the UI’s SQL tab afterwards. The plan node is AdaptiveSparkPlan with isFinalPlan=false before running and true after, and the final plan shows the strategies AQE settled on.

A left semi join returns rows from the left that have a match, with no columns from the right. A left anti join returns rows from the left with no match. Both are cheaper than a full join followed by a filter, because neither needs to carry right-side columns.

Nulls never match, so they end up on the outer side of an outer join and are dropped by an inner join. If a null should mean “unknown but joinable”, handle it explicitly with coalesce to a sentinel, and be aware that a large number of null keys concentrates into one partition, which is skew.

A join condition that Spark cannot use as an equality, for example a join on an inequality or a condition written in a where after a cross join. The plan shows CartesianProduct or BroadcastNestedLoopJoin. Check the plan whenever a join is unexpectedly slow.

Using the result of a dimension-side filter to prune partitions of the fact table at runtime, rather than scanning all of them. It applies when the fact table is partitioned on the join key and the dimension side is filtered, which is the classic star-schema shape.

Reduce before joining: filter and project both sides first. Then either bucket both on the join key so the shuffle disappears, or accept a sort-merge join and make sure the shuffle partitions are sized sensibly and skew is handled. Broadcasting is not an option at that scale.

A shuffle. Its description names the partitioning, for example hashpartitioning(city_id, 200), which tells you both the key and the target partition count. Counting Exchange nodes is the fastest way to count the shuffles in a query.

A shuffle whose output is reused by more than one branch of the plan rather than recomputed. Seeing it is good: it means Catalyst recognised a common subtree. Its absence where you expected reuse often means the two branches differ in a way you did not intend.

Aggregate and join on the same keys where possible so one shuffle serves both; repartition once by the shared key rather than letting each operator shuffle; replace distinct followed by join with a semi join; and avoid orderBy unless the output genuinely needs global ordering.

Find the bottleneck before changing anything. Open the SQL tab and the stage page, and identify the slowest stage. Then ask which of four things it is: skew, too much shuffle, spill, or too few or too many tasks. Only then change a config, and change one at a time.

Skew. The data is unevenly distributed across partitions, and a task cannot be split. Confirm it on the stage page by comparing max to median shuffle read, then handle it with adaptive skew join, a broadcast, or salting.

Raise parallelism so each task handles less, or lower the advisory partition size so AQE makes more partitions. Raising executor memory helps too, but it treats the symptom: the partitions are too large for the memory each task gets.

Usually too many small files, so the file listing and split computation dominate, or a plan grown deep by hundreds of chained withColumn calls. Check the number of input files first, then the plan size.

Too many partitions, so scheduling overhead dominates the work. Lower spark.sql.shuffle.partitions or let AQE coalesce, and coalesce before writing. The rule of thumb is partitions in the low hundreds of megabytes, not kilobytes.

Favour several medium executors over a few huge ones or many tiny ones. Around 4 to 5 concurrent tasks per executor is a common starting point, because tasks contend for the same HDFS client and because a very large heap makes full garbage collections slow. Then size memory so each task’s share covers your largest partition, and leave headroom for overhead.

Pick one. Setting both --num-executors and dynamic allocation is contradictory, and the static value only acts as an initial size. Use static sizing for predictable batch jobs, and dynamic allocation for bursty or interactive workloads, remembering that it needs shuffle tracking or an external shuffle service.

The cluster manager killed the JVM for using more than its requested total, which is heap plus overhead. It is usually overhead: off-heap buffers, the Python worker, or native libraries. Raise spark.executor.memoryOverhead rather than heap, and check whether one skewed partition is the real cause.

The default timeout for network interactions, 120s. Raise it only when you have evidence of legitimately slow operations, such as very large shuffle fetches; otherwise a timeout is reporting a dead or overwhelmed executor and raising it just delays the error.

Add -verbose:class through spark.driver.extraJavaOptions or spark.executor.extraJavaOptions. It prints which JAR every class was loaded from, which turns a guess about shading into a direct answer. Then fix it with spark.jars, a shaded assembly, or spark.driver.userClassPathFirst where appropriate.

Usually a plan deep enough that a recursive walk over it exhausts the thread stack: hundreds of columns, deeply nested structs, or a long iterative lineage. Raising -Xss on the side that threw resolves many cases; checkpointing to truncate lineage is the structural fix.

Write a log4j2.properties, ship it with --files, and point both sides at it:

spark-submit \
  --files /etc/spark/conf/log4j2.properties \
  --conf "spark.driver.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties" \
  --conf "spark.executor.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties" \
  app.jar

Because --files puts it in each working directory, the value is a bare filename, not a path.

The Environment tab of the UI lists the configs in force, which is the authoritative answer. spark.conf.get("key") works for SQL configs at runtime. A config set after the relevant component started, such as a driver JVM option in client mode, will silently not apply.

The first governs how input files are split into partitions on read, default 128MB. The second governs how many partitions a shuffle produces, default 200. Tuning read parallelism and shuffle parallelism are separate problems and people often reach for the wrong one.

repartition by the same columns you partitionBy, so each output partition is written by one or a few tasks instead of all of them. That converts thousands of small files into a handful of properly sized ones, which also makes every later read faster.

Are you reading more columns or rows than you need; is there an accidental Cartesian or a missed broadcast in the plan; are there small files; is there skew; is something cached that should not be. Each of those outweighs most config changes.

Change one thing, run on representative data at representative scale, and compare the stage that you expected to change rather than total wall clock on a warm cache. Timings from a laptop or a demo dataset are dominated by JVM warm-up and page cache, so they do not transfer.

A stream processing model where a stream is treated as an unbounded table and a query over it is re-run incrementally as data arrives. You write the same DataFrame operations as for batch, and the engine maintains the incremental state.

DStreams are RDD-based micro-batches with their own API. Structured Streaming is DataFrame-based, goes through Catalyst, supports event-time processing and watermarks, and provides end-to-end exactly-once with replayable sources and idempotent sinks. New work should use Structured Streaming.

complete is only valid with aggregations, and it rewrites everything each trigger, so it does not scale to large result tables.

It controls when a micro-batch runs. The choices are the default, which starts the next batch as soon as the previous finishes; a fixed processing-time interval; availableNow, which processes all available data and stops; and continuous processing, which is experimental and limited.

Event time is when the record says the event happened; processing time is when the engine saw it. Event time is what correctness depends on for windowed aggregation, and it is why a record arriving late must still be attributed to its original window.

It is a threshold on event time that tells the engine how late data may be. It does two things: it finalises windows so their results can be emitted in append mode, and it lets the engine drop the state for windows that can no longer change. Without one, a stateful query’s state grows without bound.

(events
  .withWatermark("started_at", "10 minutes")
  .groupBy(window(col("started_at"), "5 minutes"), col("city_id"))
  .agg(count("*")))

Essentially nothing. withWatermark on a simple pass-through that keeps no state has nothing to finalise and no state to drop. Seeing one there is a sign the author copied it from an aggregation example.

A required location where the engine records offsets processed and the state it holds, so a restarted query resumes exactly where it stopped. Every streaming query needs checkpointLocation, and you cannot share one between two queries.

By combining a replayable source that can be re-read from a recorded offset with an idempotent sink that will not duplicate on retry, plus checkpointed offsets. It is a property of the whole pipeline, not of the engine alone: an at-least-once sink gives at-least-once end to end.

A sink that hands you each micro-batch as a normal DataFrame, so you can use batch-only operations such as MERGE into a lakehouse table, or write to several destinations. It is the standard way to drive an upsert from a stream.

It depends on the change. Adding a column or changing a filter is usually safe. Changing the aggregation keys, the output mode, or the nature of the state is not, because the checkpointed state no longer matches the query, and the restart fails or must start from a fresh checkpoint.

Set the watermark to the lateness you are willing to tolerate. Data later than that is dropped from stateful aggregations. If you need it, capture it separately, for example by writing the raw stream as well and reconciling in batch.

Any operation whose result depends on data across triggers: windowed aggregation, deduplication, stream-stream joins, and arbitrary state via flatMapGroupsWithState or the newer transform-with-state API. All of them need a watermark to bound state.

Both sides must buffer rows waiting for a match, so both need watermarks and the join needs a time constraint, or state grows without bound. Outer joins additionally cannot emit a non-match until the watermark proves no match can arrive.

query.lastProgress and query.status give per-batch metrics, including input rows per second, processing rate, and state size. The two numbers that matter are whether processing rate keeps up with input rate, and whether state size is growing without bound, which is the signature of a missing or too-generous watermark.

--master yarn with --deploy-mode client or cluster. Spark starts an ApplicationMaster which requests containers through YarnAllocator, and each container becomes an executor. HADOOP_CONF_DIR or YARN_CONF_DIR must point at the cluster config so Spark can find the ResourceManager.

--master k8s://https://host:port. The driver runs in a pod and its own ExecutorPodsAllocator creates executor pods through the API server, so there is no Spark-specific cluster-manager daemon. You supply a container image and a service account with permission to create pods.

Across applications, the cluster manager arbitrates, using queues on YARN or namespaces and quotas on Kubernetes. Within one application, spark.scheduler.mode set to FAIR with pools shares slots between concurrent jobs, which is what you want for a shared notebook or JDBC server.

Highest to lowest: values set on SparkConf in code, then flags to spark-submit, then spark-defaults.conf. A properties file can be named with --properties-file, and --load-spark-defaults makes Spark read spark-defaults.conf as well.

--packages for Maven coordinates with transitive resolution, --jars for local jars on driver and executor classpaths, and --py-files for Python .zip, .egg or .py. For repeatability in production, bake dependencies into an assembly jar or a container image rather than resolving at submit time.

The application UI dies with the application. Enable event logging with spark.eventLog.enabled and a log directory, and the history server replays those logs so you can investigate a job after it finished. Without it, post-mortem analysis is guesswork.

Authentication between components with spark.authenticate and a shared secret, encryption in transit for both RPC and shuffle, Kerberos for Hadoop delegation tokens, and cluster-manager-level isolation. Avoid secrets in extraJavaOptions, since the Environment tab of the UI displays configs.

JVM proxy properties on both sides, for example -Dhttp.proxyHost, -Dhttp.proxyPort and -Dhttp.nonProxyHosts, passed through spark.driver.extraJavaOptions and spark.executor.extraJavaOptions. It is the same delivery mechanism as any other JVM flag, which is why learning one teaches the rest.

Stage-level task duration skew, shuffle spill, GC time per executor, executor loss and its cause, and for streaming the gap between input and processing rate plus state size. Alert on the structural signals rather than on total job duration, which moves for many unrelated reasons.

Nothing changed in the code, so something changed in the data or the cluster.

1. Compare the physical plan with a previous run. A join that used to broadcast and now does not is the most common cause, because the dimension table grew past the threshold.
2. Check input size and file count. Ten times the files with the same bytes changes planning cost and partition count.
3. Check skew: max versus median task time on the slowest stage. A new hot key concentrates work.
4. Check the executors tab for loss and retries, and for GC time.
5. Only then look at configs, which did not change.

The usual answer to this scenario is a lost broadcast or a newly skewed key.

Classic skew. Confirm on the stage page that the long tasks also have far larger shuffle read. Then:

• Check whether AQE skew handling could fire. It needs a partition larger than 5 times the median and larger than 256 MB, so on a smaller job it never triggers even with terrible skew.
• If one side is small, broadcast it and remove the shuffle entirely.
• Otherwise salt the hot key, or split the hot keys out and union the results.

Do not enable speculation for this: the duplicate task is equally slow and just wastes a slot.

Because output parallelism times partition count equals file count: 200 shuffle partitions writing 250 partition values gives 50,000 files. Fix it by repartitioning on the same columns you partition by, so each value is written by one task:

(df.repartition("city_id")
   .write.partitionBy("city_id")
   .mode("overwrite").parquet(path))

Also consider whether the partition column is too high-cardinality for partitioning at all, and cap file size with spark.sql.files.maxRecordsPerFile.

Something is pulling data back. Look for collect(), toPandas(), a broadcast of a side that is not small, or show() on a very wide result. Also consider task count: a job with hundreds of thousands of tasks makes the driver track a lot of metadata. spark.driver.memory defaults to 1g, which is a coordinator’s budget, not a data-processing budget.

1. Confirm it is the container total, not the heap: the message comes from the cluster manager, not a JVM OutOfMemoryError.
2. Raise spark.executor.memoryOverhead first, because off-heap buffers, the Python worker and native libraries live there.
3. Check for skew, since one huge partition in one task is a common root cause that looks like a memory shortage.
4. Raise partition count so each task’s share is smaller.
5. Only then raise executor memory.

Raising memory first often just moves the failure later.

Almost never the fetch settings. The executor that held those shuffle blocks is gone, was killed for memory, or is too busy to serve. Investigate why it died. An external shuffle service makes shuffle output outlive the executor, which is also what dynamic allocation needs to release executors safely.

3 GB is far past the 10 MB broadcast threshold, so a broadcast is out unless you can reduce it. In order:

1. Filter and project the dimension side first. If the query only needs two columns and a subset of rows, 3 GB may become broadcastable.
2. If not, accept a sort-merge join and size the shuffle sensibly.
3. If this join runs repeatedly, bucket both tables on the join key with the same bucket count, which removes the shuffle permanently.
4. Handle skew on the join key separately.

Forcing a broadcast hint on 3 GB moves the failure to the driver, which must collect it first.

A stateful operation without an effective watermark, so nothing is ever old enough to drop. Check that the watermark column is event time from the payload and not the source’s ingest timestamp, that the watermark is actually attached before the aggregation, and that the allowed lateness is not enormous. Confirm with lastProgress, where state rows should plateau rather than climb.

Silent wrong results point at semantics, not performance.

• An inner join dropping non-matching rows, including rows whose key is null.
• union matching columns by position where the two sides differ in order. Use unionByName.
• A filter written against a column that is now sometimes null, since col = value is never true for null.
• On an upgrade to Spark 4, ANSI mode now raising or changing results where invalid casts previously produced null.

Compare row counts at each stage of the pipeline to find where they fall off.

A window function is the idiomatic way:

from pyspark.sql import Window
from pyspark.sql.functions import row_number, desc, col

latest = (changes
    .withColumn("rn", row_number().over(
        Window.partitionBy("trip_id").orderBy(desc("updated_at"))))
    .where(col("rn") == 1).drop("rn"))

It shuffles once by key. If the key is skewed, that single partition is your bottleneck, so check the distribution first. For a stream, dropDuplicates with a watermark is the bounded-state equivalent.

The shape that works at one scale rarely fails at another for a config reason; it fails because partition sizes scaled with the data.

1. Check partitions: total bytes divided by partition count should land in the low hundreds of megabytes.
2. Lower spark.sql.adaptive.advisoryPartitionSizeInBytes or raise the shuffle partition count so tasks are smaller.
3. Look for an aggregation with far higher cardinality at scale than in the sample.
4. Check whether something cached at sample size no longer fits and is now evicting execution memory.

The filter is not pruning. Either the table is not partitioned on the date column, or the predicate is written so it cannot be pushed down, for example by wrapping the column in a function such as to_date(ts) = '...' instead of comparing the partition column directly. Check the plan for PartitionFilters and PushedFilters: if the date predicate appears in neither, nothing is being skipped.

Plain Parquet directories have no commit protocol, so two concurrent writers can interleave and a failure can leave partial output. Either serialise the writers, or move to a table format with atomic commits: Hudi, Iceberg or Delta. That is the structural fix, and it is why lakehouse formats exist.

Ask first whether the aggregation is needed at that cardinality.

• If it feeds a join, aggregate after joining on a coarser key.
• If it feeds an approximate answer, use approx_count_distinct rather than exact countDistinct, which is a second shuffle.
• If output is written partitioned by that column, the cardinality is also a small-files problem.
• Make sure the shuffle partition count matches the cardinality’s scale so partitions are neither tiny nor huge.

Separate the engine from the table format. Spark is the compute; Hudi and Iceberg are table formats with different strengths, so the questions are about write pattern and read reach:

• Are the writes keyed upserts at high frequency? Hudi has a record index for that; Iceberg plans a join.
• Which engines must read it? Iceberg has the widest warehouse support.
• Who operates the table services, compaction and cleaning, in each case?

If the answer is “both, for different readers”, Apache XTable converts metadata in place without copying data.