Spark Performance Tuning: Common Parameters, Programming Tips, Shuffle and Join Optimization

Spark is a powerful engine for big‑data analytics, and the author shares practical tuning experience to improve job performance.

1. Common Configuration Parameters

--driver-memory 4g : driver memory size, typically 4g (6g‑12g if broadcast variables are used)
--executor-memory 4g : memory per executor, increase to 6g for large data sets
--num-executors 15 : total number of executors, scale up for hundreds of gigabytes or terabytes
--executor-cores 2 : cores per executor, determines parallel tasks per executor
--spark.default.parallelism 200 : default parallelism, usually 2‑3×(num‑executors × executor‑cores)
--spark.storage.memoryFraction 0.6 : max fraction of executor memory for cached RDDs
--spark.shuffle.memoryFraction 0.2 : memory fraction for shuffle aggregation, excess spills to disk
--spark.yarn.executor.memoryOverhead 1G : extra memory reserved for executor overhead

2. Spark Programming Recommendations

Avoid creating duplicate RDDs; reuse existing data whenever possible.

Minimize use of shuffle‑heavy operators (e.g., reduceByKey, join, distinct, repartition) and prefer map‑type transformations.

Prefer aggregateByKey or reduceByKey over groupByKey to reduce network and disk I/O.

Use repartition for RDD[V] and partitionBy for RDD[K,V].

Replace plain map with mapPartitions, and foreach with foreachPartitions.

Apply coalesce after filter to reduce the number of partitions.

Cache long‑running RDDs; enable Kryo serialization or set storage level to MEMORY_AND_DISK_SER for large partitions.

Break complex transformations into smaller, composable steps.

When unioning many RDDs, use SparkContext.union(Array(RDD)) instead of nested rdd.union calls.

Specify partition numbers directly in group/join/…ByKey operations to avoid extra repartitioning.

Ensure each task processes >128 MB of data per stage.

For joins where one side is small, use Broadcast Join.

For Cartesian products, pass the smaller RDD as the argument (e.g., bigRDD.cartesian(smallRDD)).

When broadcasting large lookup tables, increase executor‑cores and executor‑memory accordingly.

If an object is too large to broadcast, use zipPartitions with a hash‑join approach.

After repartition, use repartitionAndSortWithinPartitions for combined shuffle‑and‑sort.

3. Shuffle Performance Optimization

Shuffle consists of a write phase (intermediate data spilled to local disks) and a read phase (tasks pull required partitions from other executors). Reducing shuffle volume, choosing an appropriate ShuffleManager (hash, sort, tungsten‑sort), and enabling bypass or file consolidation can significantly improve performance.

- spark.shuffle.file.buffer : default 32M, increase (e.g., 64M) if memory permits.
- spark.shuffle.io.maxRetries : default 3, increase for large shuffle data.
- spark.shuffle.manager : choose hash, sort, or tungsten‑sort.
- spark.shuffle.sort.bypassMergeThreshold : increase to enable bypass when sort manager is used.
- spark.shuffle.consolidateFiles : set true with HashShuffleManager to merge output files.

4. Join Performance Optimization

Join is often the most expensive operation. Strategies include:

Broadcast Join for small tables.

Shuffle Hash Join when both tables are large, partitioning by join key.

Sort‑Merge Join for two large tables, involving shuffle, sort, and merge phases.

Ensure both RDDs share the same partitioner to avoid unnecessary shuffle.

Remove duplicate keys before joining.

5. Data Skew Mitigation

Detect skewed keys via countByKey or SQL analysis, then apply techniques such as filtering hot keys, increasing spark.sql.shuffle.partitions, using two‑stage aggregation (random prefix + global aggregation), or replicating the small side of the join.

6. Common Error Summaries and Fixes

java.lang.OutOfMemoryError: unable to create new native thread – increase OS max process limit.
No Space Left on device (shuffle temp files) – set SPARK_LOCAL_DIRS to a larger disk.
Worker work directory fills up – schedule periodic cleanup.
Executor Lost due to OOM or data skew – increase executor memory or raise partition count.
MetadataFetchFailedException / FetchFailedException – check shuffle manager settings and network stability.

By applying the above configuration tweaks, programming best practices, and targeted optimizations for shuffle, join, and skew, Spark jobs can achieve more stable execution and better resource utilization.