Spark Performance Tuning and Data Skew

This is where senior candidates are separated from mid-level ones. Anyone can say 'add more executors.' You should be able to say which metric you looked at, what it told you, and why your fix addressed the actual bottleneck.

Step 1 — Diagnose before you tune

Step 2 — Kill the skew

Skew is the number-one cause of 'my job hangs at 199/200 tasks.' In telecom data it is everywhere: one busy cell site, one default unknown device id, one null subscriber key holding millions of rows.

# 1. CONFIRM the skew — never guess
(df.groupBy("cell_site_id")
   .count()
   .orderBy(F.col("count").desc())
   .show(20))

# 2a. Easiest fix: let AQE handle it (Spark 3+)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# 2b. Broadcast the small side so there is no shuffle at all
result = big_df.join(F.broadcast(small_dim), "cell_site_id")

# 2c. Salting — when one key is genuinely enormous
SALT = 16
left = big_df.withColumn("salt", (F.rand() * SALT).cast("int"))
right = (small_df
    .withColumn("salt_arr", F.array([F.lit(i) for i in range(SALT)]))
    .withColumn("salt", F.explode("salt_arr"))
    .drop("salt_arr"))

joined = left.join(right, ["cell_site_id", "salt"]).drop("salt")

Step 3 — Pick the right join strategy

# Raise the broadcast threshold when your dimension is bigger than 10 MB
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 200 * 1024 * 1024)

# Verify what Spark actually chose — do this, do not assume
joined.explain(mode="formatted")   # look for BroadcastHashJoin vs SortMergeJoin

Step 4 — Fix the small-file problem

Streaming jobs and over-partitioned batch writes create millions of tiny files. Every file costs a listing call and a task, so reads get slower and slower over time.

• Target file sizes of roughly 128 MB–1 GB
• coalesce() before writing, or set spark.sql.adaptive.coalescePartitions.enabled
• Run a compaction job that rewrites yesterday's small files into big ones
• Do not over-partition on disk — partitioning by date/hour/site when a site has 10 rows an hour is the cause, not the cure

Step 5 — Memory and resource sizing

# A sane starting point for a large batch job
--executor-memory 16g          # heap per executor
--executor-cores 4             # 3-5 is the sweet spot for HDFS/GCS throughput
--num-executors 40
--conf spark.memory.fraction=0.6            # execution + storage share
--conf spark.sql.shuffle.partitions=800     # ~= 2-3x total cores
--conf spark.sql.adaptive.enabled=true

Practice questions