Learn Apache Spark - Performance Tuning & Optimization

#Introduction

Correct Spark code isn't necessarily fast. In episode 9 you'll learn that Spark's performance is largely determined by configuration and an understanding of how the engine works. Two queries that produce the same output can differ by dozens of times in duration, depending on how Spark is asked to execute them.

Why does this matter? Because in production, cluster resources aren't free and time is a cost. An ETL job that runs for two hours instead of twenty minutes means a big waste. Engineers who can optimize Spark not only save money, they also shorten the wait time of the entire data team.

This episode covers four pillars of optimization: tuning the basic configuration, broadcast joins and caching, reading the Catalyst execution plan, and strategies for dealing with data skew and shuffle overload.

#Tuning the Basic Configuration

#Memory Configuration

Executor memory is divided into several regions. The two most important: execution memory for operations like joins and shuffle, and storage memory for caching. Both share a single pool that can borrow from each other, controlled by spark.memory.fraction:

spark-submit --executor-memory 8g --conf spark.memory.fraction=0.6 job.py

A practical rule: leave room for JVM overhead and the system. If an executor runs out of memory, Spark spills data to disk and the job slows down drastically. Make sure --executor-memory is balanced with the number of cores so you're not paying for idle resources.

#Shuffle Configuration

Shuffle is the biggest cost. Two parameters that are often tuned:

• spark.sql.shuffle.partitions: the number of output partitions after a shuffle — 200 by default.
• spark.shuffle.compress: compression of shuffled data — leave it true.

spark-submit --conf spark.sql.shuffle.partitions=100 job.py

The ideal value depends on the number of cores and data size. Too few makes each task heavy; too many creates lots of small tasks whose overhead outweighs the benefit.

#Parallelism

Parallelism is determined by the number of partitions, not the number of cores. A general rule: aim for 2-3 partitions per core for CPU-bound workloads, and more for I/O-bound workloads because tasks will be waiting on network or disk. Inspect the current parallelism with:

Viewing the number of partitions

df.rdd.getNumPartitions()

#Broadcast Joins and Caching

#Broadcast Joins

If one side of a join is small (usually under 10MB by default, settable up to 200MB with autoBroadcastJoinThreshold), Spark can send a copy to every executor. This way the join happens locally with no shuffle at all:

Explicit broadcast join

from pyspark.sql import functions as F
 
dimensi = spark.read.parquet("data/dim_kota.parquet")
hasil = transaksi.join(F.broadcast(dimensi), "kota_id")

F.broadcast(dimensi) tells Catalyst to treat dimensi as a broadcast relation. For joins comparing a giant transaction table with a small dimension table, this technique can cut time by up to dozens of times.

#Caching and Persistence

Data used repeatedly within one application should be cached so it isn't recomputed from the source:

Cache and persist

df.persist()
print(df.count())
print(df.filter("stok > 0").count())
df.unpersist()

Without persist(), every action triggers a full recomputation from the source. But remember: caching consumes storage memory. Only cache data that's genuinely reused, and call df.unpersist() when done to release the memory.

#Partition Tuning

When partitions are unbalanced or too numerous, adjust with repartition() or coalesce():

Changing the number of partitions

df = df.repartition(200, "kota_id")
df = df.coalesce(8).write.parquet("data/hasil.parquet")

repartition triggers a full shuffle and is useful when increasing parallelism; coalesce merges partitions without a full shuffle and is useful when reducing the number of files before writing.

#Catalyst and the Execution Plan

#Reading the Physical Plan

The best optimization starts with diagnosis. df.explain() shows how Spark will execute the query:

Viewing the execution plan

df.groupBy("kota").agg(F.sum("jumlah")).explain("extended")

The output shows the parsed logical plan, analyzed plan, optimized plan, up to the physical plan. Watch for three things when reading it:

• ScanFilter: whether the filter is applied as early as possible at the data source.
• Exchange: nodes that indicate a shuffle — the fewer, the better.
• BroadcastHashJoin vs SortMergeJoin: the join strategy Catalyst chose.

#Optimizations That Are Already Automatic

Catalyst does many things without your intervention:

• Predicate pushdown: filters are pushed to the data source so fewer files are read.
• Projection pruning: unused columns are never read.
• Join reordering: join order is chosen based on statistics.
• Codegen: generates JVM bytecode specific to a given query.

Your job isn't to replace Catalyst, but to make sure the data gives it the right signals — for example providing accurate table statistics so join reordering works well.

#Avoiding Skew, Shuffle Overload, and Expensive Operations

#Data Skew

Data skew happens when one partition handles far more data than the others — for example one city holding 90 percent of transactions. The symptom: one task takes a long time while the others finish quickly. Common solutions:

• Salting: add a random prefix to the hot key to spread the load.
• Broadcast join: for hot keys with a small side.
• Adaptive Query Execution (AQE): enable spark.sql.adaptive.enabled=true, which automatically handles skew joins in Spark 3+.

spark-submit --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.skewJoin.enabled=true job.py

#Shuffle Overload

Every join, groupBy, and reduceByKey triggers a shuffle. To reduce the load:

• Replace groupByKey with reduceByKey or partial aggregations.
• Use broadcast joins for the small side.
• Filter and prune columns before the join, not after.
• Avoid collect() on large data — it sends all the data to the driver and exhausts driver memory.

#Expensive Operations

Some operations should be avoided or redesigned:

collect()     → send all data to the driver
cartesian()   → explodes exponentially
groupByKey()  → shuffles all raw values
row-level udf → no pushdown, far slower than Spark expressions

Especially for UDFs: if it can be expressed with the built-in pyspark.sql.functions, use that. Python UDFs force data to cross the JVM-Python boundary and eliminate Catalyst optimizations.

#Conclusion

Episode 9 rounds out your optimization skills: memory and shuffle tuning set the resource limits, broadcast joins and caching reduce repetitive work, reading the physical plan is the diagnostic skill, and avoiding skew and shuffle overload is the art of keeping pipelines balanced.

Key takeaways:

• Start tuning from observation, not guesses: use explain() and the Spark UI.
• Broadcast joins are the first weapon for joins with a small side.
• Cache only data that's reused, and release it with unpersist().
• Enable Adaptive Query Execution to handle skew automatically.
• Avoid collect(), cartesian(), and groupByKey on large data.

In the next episode, episode 10, we'll discuss Spark Streaming and Structured Streaming — the difference between DStreams and Structured Streaming, building streaming pipelines with readStream and writeStream, trigger modes, watermarking, and output modes, plus sinks to Kafka, files, console, and storage.