TopolearnExpert Spark work is mostly diagnosis: finding the slow stage, the skewed task or the spilling shuffle, and fixing the right setting.
The Spark UI (port 4040 for a running app, or the History Server afterwards) is the main diagnostic tool:
- Jobs and Stages: find the stage that takes the time. Stage boundaries are shuffles.
- Task summary: compare max with median task duration and input size. A max many times the median means skew.
- Spill (memory/disk): tasks ran out of execution memory and wrote to disk; partitions are too large.
- Shuffle read/write sizes: how much data crosses the network.
- SQL tab: the executed plan with row counts and timings per operator, including AQE changes.
- Executors: GC time, memory, failed tasks.
Executor memory model: spark.executor.memory is the JVM heap. After about 300 MB is reserved, spark.memory.fraction (0.6) forms a unified region shared by execution (shuffles, joins, sorts, aggregations) and storage (cache); execution can evict cached blocks. The rest holds user data structures. spark.executor.memoryOverhead covers off-heap and native memory, and Python workers need spark.executor.pyspark.memory or enough overhead, or containers get killed.
Tuning priorities: fix the data first (pruning, filter early, avoid skew, right file sizes), then partitioning (advisoryPartitionSizeInBytes, maxPartitionBytes), then join strategy, and only then resources: typically 4 to 5 cores per executor, with memory per core sized to partition size.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
skewed = spark.range(400_000).withColumn(
"k", F.when(F.col("id") < 300_000, 0).otherwise(F.col("id") % 100))
per_part = (skewed.repartition(8, "k")
.withColumn("pid", F.spark_partition_id())
.groupBy("pid").count().orderBy(F.desc("count")).collect())
sizes = [r["count"] for r in per_part]
print(sizes[0], sorted(sizes)[len(sizes) // 2]) # one huge partition vs a small median
print(sizes[0] > 10 * sorted(sizes)[len(sizes) // 2]) # True -> skew
salted = (skewed.withColumn("salt", (F.rand(seed=7) * 8).cast("int"))
.repartition(8, "k", "salt")
.groupBy(F.spark_partition_id().alias("pid")).count().collect())
print(max(r["count"] for r in salted) < sizes[0]) # True: work spread out
print(spark.conf.get("spark.memory.fraction", "0.6")) # 0.6Remember
- Find the slow stage, then compare max vs median tasks to spot skew
- Spill means partitions too large for execution memory
- Unified memory: execution and storage share
spark.memory.fractionof the heap
Watch out for
- Raising executor memory to hide skew or oversized partitions instead of fixing the data layout.