Apache Spark Architecture Questions

Covers core Apache Spark architecture and programming model, including the roles of the driver and executors, cluster manager options, resource allocation, executor memory and cores, partitions, tasks, stages, and the directed acyclic graph used for job execution. Explains lazy evaluation and the distinction between transformations and actions, fault tolerance mechanisms, caching and persistence strategies, partitioning and shuffle behavior, broadcast variables and accumulators, and techniques for performance tuning and handling data skew. Compares Resilient Distributed Datasets, DataFrames, and Datasets, describing when to use each API, the benefits of the DataFrame and Spark SQL APIs driven by the Catalyst optimizer and Tungsten execution engine, and considerations for user defined functions, serialization, checkpointing, and common data sources and formats.

EasyTechnical

22 practiced

Explain broadcast variables and accumulators in Spark. Describe their lifecycle, typical use-cases, and limitations (size constraints, update semantics, type restrictions). Provide an example scenario where a broadcast variable significantly improves a join performance.

Sample Answer

Definition & purpose:- Broadcast variables: read-only copies of a value cached on each executor to avoid shipping large read-only data repeatedly (e.g., small lookup/dimension tables).- Accumulators: write-only variables tasks can increment (counters, sums). Driver can read final value after action for metrics/debugging.

Lifecycle:- Broadcast: created on the driver (sc.broadcast(obj)), serialized and sent to executors on first use, cached in executor memory; lives until explicitly unpersisted (.unpersist()) or SparkContext stops.- Accumulator: created on driver (acc = sc.longAccumulator()), tasks update it during job. Updates are sent to driver; final value available after job completes. Be careful: task retries may double-count.

Typical use-cases:- Broadcast: small dimension table for joins, configuration maps, ML model parameters.- Accumulators: counting processed records, logging bad rows, progress metrics during ETL.

Limitations & semantics:- Size constraints: broadcast should be relatively small — typically tens to a few hundreds of MB depending on cluster memory; very large broadcasts cause driver/executor OOMs and heavy network overhead.- Update semantics: broadcasts are immutable on executors (read-only). Accumulators support only task-side updates; driver reads only after actions. Accumulators may overcount if tasks are retried.- Type restrictions: broadcasts can contain arbitrary serializable objects (but ensure efficient serialization). Accumulators: use built-in numeric accumulators or implement AccumulatorV2 for custom types; ensure thread-safety/serialization.- Fault/retry behavior: accumulator updates are not idempotent; design with idempotency or use metrics tolerant to duplicates.

Example where broadcast improves join:Scenario: large fact table (billions of rows) join with a small dimension (~50 MB) for enrichment. A regular join triggers a shuffle of the big table keys. If dimension fits in memory, broadcast it to avoid shuffle.

PySpark example:

python

from pyspark.sql import functions as F
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.getOrCreate()
fact = spark.read.parquet("s3://bucket/fact")
dim = spark.read.parquet("s3://bucket/dim_small")  # ~20-50 MB

# broadcast hint avoids shuffling the large fact RDD
joined = fact.join(broadcast(dim), fact.key == dim.key, "left")
joined.write.parquet("s3://bucket/enriched_fact")

Benefit: avoids expensive shuffle of the large dataset, reduces network IO and job time from hours to minutes in many cases.

EasyTechnical

27 practiced

Explain the roles and responsibilities of the Spark driver and Spark executors in a distributed Spark application. Your answer should cover where the SparkContext lives, how tasks are scheduled from the driver to executors, what metadata/state is held by the driver versus executors, how memory and cores are allocated to each, and the observable failure modes when a driver or an executor fails in production.

Sample Answer

Brief answer:

The Spark driver is the single JVM process that runs the SparkContext (and SparkSession). It is responsible for: accepting the user program, building the logical plan and RDD/DataFrame DAG, converting jobs to stages (DAGScheduler), scheduling tasks (TaskScheduler) via the cluster manager (YARN/Kubernetes/Standalone), holding cluster-level metadata (job/stage/task state, accumulators, broadcast variable metadata) and coordinating retries. The driver also holds the REPL/interactive session state and driver-side caches (references to Broadcast variables).

Executors are worker JVM processes launched on cluster nodes. Each executor:- Runs tasks assigned by the driver- Holds working memory for task execution (task memory + execution memory + storage memory for cached RDDs/DataFrames)- Stores cached blocks and shuffle files (local disk)- Reports heartbeat and task status back to the driver

Scheduling & resource allocation:- SparkContext lives in the driver; it requests executors from the cluster manager.- Resources configured via spark.driver.memory, spark.driver.cores, spark.executor.memory, spark.executor.cores, and spark.executor.instances (or dynamic allocation).- The driver’s TaskScheduler maps stage tasks to executors considering locality; executors run tasks up to the number of cores allocated.

Metadata/state:- Driver: DAG, job/stage/task metadata, task scheduler state, broadcast metadata, accumulators, job listener hooks.- Executors: cached partitions, shuffle files, in-memory buffers for tasks, task-local state.

Failure modes and observables:- Driver failure: application-level failure. Symptoms: entire Spark app dies, no new tasks scheduled, client loses connection; in cluster mode the driver restart depends on cluster manager (YARN can restart if configured, but running job may still be lost). Mitigation: enable driver high-availability (where supported), checkpoint critical state, run in cluster mode with automatic restarts.- Executor failure: individual tasks fail and are retried on other executors; symptoms include repeated task failures, increased task latency, stage retries, missing shuffle files (if external shuffle service not used) causing stage failures. Common causes: executor OOM (logs show Java OOM or GC pressure), disk full, network issues. Mitigation: tune executor memory/cores, enable external shuffle service (preserve shuffle on executor death), use dynamic allocation, increase retries (spark.task.maxFailures), and add monitoring/alerts.

Key operational tips:- Set executor memory and cores to balance parallelism and GC overhead (e.g., 1 core per task slot is common).- Use spark.dynamicAllocation with an external shuffle service to scale executors safely.- Monitor driver heap and GC; driver OOMs are critical. Monitor executor OOMs and shuffle I/O.- Use durable checkpoints for long lineage to reduce recomputation after failures.

EasyTechnical

31 practiced

In PySpark, demonstrate how to persist a DataFrame using an appropriate StorageLevel, ensure it survives transient executor failures if possible, and show how to properly unpersist it. Include a short code example and explain when to choose MEMORY_ONLY, MEMORY_AND_DISK, or serialized storage.

Sample Answer

Approach: choose a StorageLevel that balances speed, memory, and fault-tolerance; persist the DataFrame, use a replicated storage level if you want resilience to executor loss, and unpersist when done.

python

from pyspark.sql import SparkSession
from pyspark import StorageLevel

spark = SparkSession.builder.appName("persist-example").getOrCreate()

df = spark.read.parquet("/data/events/")

# Persist with replication so cached partitions survive a transient executor failure
df.persist(StorageLevel.MEMORY_AND_DISK_2)  # replicate blocks on 2 executors

# Use the cached dataframe
df.count()   # triggers materialization
df.show(5)

# When finished, remove from cache (block until unpersist completes if desired)
df.unpersist(blocking=True)

spark.stop()

Key points / when to choose:- MEMORY_ONLY: fastest; keep objects as JVM objects in RAM. Use if dataset fits in memory and you need fastest access. If a partition can't be cached it will be recomputed (no disk spill).- MEMORY_AND_DISK: safer when data may not fit in memory; spills partitions to disk when memory is insufficient. Good default for large datasets.- MEMORY_ONLY_SER / MEMORY_AND_DISK_SER: store serialized JVM objects (smaller footprint, less GC overhead) but slower CPU deserialization; choose when memory is tight.- Use the _2 (replicated) variants (e.g., MEMORY_AND_DISK_2) to survive single-executor failures — at cost of extra memory/disk.- Call unpersist(blocking=True) to ensure resources are freed before continuing.

MediumTechnical

30 practiced

You have a join between a very large fact table and a much smaller dimension table but your job still does massive shuffles and runs slowly. Describe how to detect data skew and three concrete techniques to mitigate skew (code-level and config-level) to speed up the join.

Sample Answer

Detecting skew- Compute key frequency on the large fact table and inspect heavy hitters:

fact.groupBy("join_key").count().orderBy(desc("count")).limit(20).show()

- Observe shuffle/read metrics in Spark UI: a few tasks with very large shuffleRead or long duration indicate skew.- Check partition sizes: rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect() shows uneven counts.

Three concrete mitigation techniques

1) Broadcast the small dimension (map-side join)When the dimension is truly small (< spark.sql.autoBroadcastJoinThreshold or manually set), broadcast eliminates shuffle.

from pyspark.sql.functions import broadcast
joined = fact.join(broadcast(dim), "join_key")

Config: spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10*1024*1024) or use hint: fact.join(dim.hint("broadcast"), "join_key")

2) Salting hot keysIf a few keys dominate, add a random salt to both sides for hot keys so their work is spread across partitions, then aggregate to remove salts.

from pyspark.sql.functions import when, rand, floor, concat, lit
num_salts = 10
fact2 = fact.withColumn("salt", when(col("join_key").isin(hot_keys), floor(rand()*num_salts)).otherwise(lit(0)))
fact2 = fact2.withColumn("join_key_salted", concat(col("join_key"), lit("_"), col("salt")))
dim2 = dim.withColumn("salt", when(col("join_key").isin(hot_keys), explode(array([lit(i) for i in range(num_salts)]))).otherwise(lit(0)))
dim2 = dim2.withColumn("join_key_salted", concat(col("join_key"), lit("_"), col("salt")))
joined = fact2.join(dim2, "join_key_salted")
# then aggregate to remove salt

Reason: spreads heavy-key rows across num_salts partitions.

3) Repartition / skew-aware partitioning and adaptive execution- Repartition the fact table by join key to balance partitions: fact.repartition(200, "join_key")- Use Spark Adaptive Query Execution (AQE) to automatically handle skewed partitions:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

AQE will split skewed partitions at runtime. Also tune shuffle partitions: spark.conf.set("spark.sql.shuffle.partitions", "500")

Trade-offs and notes- Broadcasting uses memory on executors; ensure dim fits.- Salting increases data volume and requires post-aggregation.- Repartitioning costs a shuffle; choose when join is repeated downstream.- Use Spark UI to verify improvements (reduced max task time, more even shuffleRead).

EasyTechnical

32 practiced

Describe the roles of the Catalyst optimizer and Tungsten execution engine in Spark SQL. Explain in plain terms what they optimize (logical rules, physical planning, code generation, memory layout) and how those optimizations lead to faster DataFrame queries compared to naive transformations.

Unlock Full Question Bank

Get access to hundreds of Apache Spark Architecture interview questions and detailed answers.

Join thousands of developers preparing for their dream job.