Master Spark Performance: Key Tuning, Shuffle & Join Optimization

Common Spark Parameter Settings

--driver-memory 4g : Memory for the driver process. 4 GB is sufficient for most jobs without broadcast variables; increase to 6‑12 GB when large broadcast variables are used.

--executor-memory 4g : Memory per executor. 4 GB works for typical workloads; raise to 6 GB or more for large datasets.

--num-executors 15 : Total number of executors. A few dozen is common; for hundreds of GB to TB‑scale data consider 100‑200 executors.

--executor-cores 2 : CPU cores per executor, controlling concurrent tasks per executor.

YARN clusters often enforce a total resource cap, e.g., executor-memory * num-executors < 400G . Adjust parameters to stay within this limit.

spark.default.parallelism 200 : Default parallelism for RDD jobs. If unset Spark derives it from HDFS block count, which may under‑utilize resources. A good rule of thumb is 2‑3 × (num‑executors × executor‑cores).

spark.storage.memoryFraction 0.6 : Fraction of executor memory that can be used for cached RDD data (default 0.6).

spark.shuffle.memoryFraction 0.2 : Fraction of executor memory allocated for shuffle aggregation (default 0.2). Exceeding this causes spill to disk.

spark.yarn.executor.memoryOverhead 1G : Additional off‑heap memory reserved per executor beyond executor‑memory .

Spark Programming Recommendations

Avoid creating duplicate RDDs; reuse existing ones whenever possible.

Minimize use of shuffle‑heavy operators (e.g., reduceByKey, join, distinct, repartition). Prefer map‑style transformations that do not trigger shuffle.

Prefer aggregateByKey or reduceByKey over groupByKey to perform partial aggregation locally and reduce network I/O.

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

Replace map with mapPartitions and foreach with foreachPartitions for better resource utilization.

Apply coalesce after a filter to reduce the number of partitions.

If an RDD is reused and expensive to compute, cache it. For large partitions enable Kryo serialization or use MEMORY_AND_DISK_SER storage level.

Break complex transformations into smaller composable steps (e.g., separate map and filter).

When unioning many RDDs, use SparkContext.union(Array(rdd1, rdd2, ...)) instead of nested rdd.union(...) calls to avoid deep call stacks.

Specify partition counts directly in operations like groupByKey, join, etc., rather than adding extra repartition calls.

Ensure each task processes more than 128 MB of data per stage to keep CPUs busy.

For a small‑to‑large join, use a Broadcast Join: broadcast the small RDD and join locally.

When performing a Cartesian product, pass the smaller RDD as the first argument (e.g., bigRDD.cartesian(smallRDD)).

If a large object must be broadcast, increase executor cores and memory; required memory is roughly 2 × objectSize.

When an object is too large to broadcast but key‑based lookups are needed, use zipPartitions with a hash‑join style implementation.

For sorting after repartition, prefer repartitionAndSortWithinPartitions, which combines shuffle and sort in a single step.

Shuffle Performance Optimization

What Is Shuffle?

Shuffle redistributes records that share the same key across the cluster so they can be aggregated or joined. It writes intermediate data to local disks and then transfers it over the network to the appropriate reducers.

Operations That Trigger Shuffle

Transformations that create a wide dependency trigger shuffle, e.g., repartition, groupByKey, reduceByKey, join, cogroup, distinct. Narrow‑dependency operations such as map, filter, union do not.

Shuffle Process

Shuffle Write : Each map task writes its intermediate results to local disk files, partitioned by the target reducer key. The number of output files equals numMapTasks × numReduceTasks .

Shuffle Read : Reduce tasks pull the relevant files from all map executors, buffer the data, aggregate, and then fetch the next batch. Insufficient buffer size or too few reduce tasks can cause large data pulls and JVM crashes.

Shuffle Manager Types

The spark.shuffle.manager setting selects the implementation. Options after Spark 1.5 are:

HashShuffleManager (default before Spark 1.2)

SortShuffleManager (default from Spark 1.2 onward)

tungsten‑sort : Uses off‑heap memory for higher efficiency.

Bypass Mechanism

Set spark.shuffle.sort.bypassMergeThreshold (default 200). When the number of shuffle‑read tasks is below this threshold, SortShuffleManager skips the sort step and behaves like HashShuffleManager, merging temporary files after write.

HashShuffleManager Optimizations

Enable spark.shuffle.consolidateFiles=true to merge many small shuffle files into fewer large ones, reducing disk I/O and improving read performance (10‑30 % gain over bypass).

Additional Shuffle Tuning Parameters

spark.shuffle.file.buffer (default 32 MB): Increase to 64 MB if memory permits to reduce I/O calls.

spark.shuffle.io.maxRetries (default 3): Raise when shuffle reads frequently fail.

Common Shuffle‑Related Errors

MetadataFetchFailedException : Missing shuffle output location.

FetchFailedException : Unable to connect to a remote executor during shuffle read.

FileNotFoundException in shuffle temp files: Temporary shuffle files were cleaned up prematurely.

Shuffle Error Mitigation

Filter unnecessary data before shuffle to reduce volume.

Adjust spark.sql.shuffle.partitions (default 200) to increase parallelism for large shuffles.

For RDD jobs, set spark.default.parallelism to 2‑3 × total cores.

Increase executor memory via spark.executor.memory.

Detect data skew with rdd.countByKey() or Spark SQL SELECT key, COUNT(*) FROM table GROUP BY key and apply skew‑handling strategies.

Join Performance Optimization

Spark SQL Join Strategies

Hash Join builds a hash table for the smaller table and probes it with the larger one. Two variants:

Broadcast Hash Join : Small table is broadcast to all executors; suitable when the small table fits in memory.

Shuffle Hash Join : Both tables are partitioned by the join key, then each partition performs a local hash join.

Sort‑Merge Join

Used for large‑scale joins. It consists of three steps:

Shuffle both tables by join key.

Sort each partition locally.

Merge the sorted streams, emitting matches when keys align.

Spark RDD Join Strategies

Broadcast a small RDD and perform a map‑side join.

Ensure both RDDs share the same partitioner to avoid a shuffle; the core cogroup method decides whether a shuffle is needed based on partitioner compatibility.

Deduplicate keys before join to reduce work.

Data Skew Handling

Skew occurs when a few keys dominate the data, causing some tasks to process far more records than others.

Analysis : Use Spark SQL SELECT key, COUNT(*) FROM table GROUP BY key or RDD countByKey() to identify hot keys.

Mitigation Techniques :

Pre‑aggregate or filter skewed keys at the source (e.g., Hive).

Drop negligible skewed keys if they do not affect results.

Increase spark.sql.shuffle.partitions to raise parallelism.

Apply two‑stage aggregation: add a random prefix to keys, perform local aggregation, then remove the prefix for global aggregation.

For a skewed join, split the hot‑key data into a separate RDD, broadcast it, and join the remaining data after randomizing key distribution.

Other Common Errors and Fixes

OutOfMemoryError – Unable to Create New Native Thread

The OS limit on the number of threads is reached, not a lack of heap memory. Increase the maximum number of processes/threads (e.g., ulimit -u or edit /etc/security/limits.conf).

No Space Left on Device (Shuffle Temp Files)

Shuffle writes fill the /tmp directory (often a tmpfs mount). Redirect Spark’s local directories to a larger disk:

export SPARK_LOCAL_DIRS=/home/utoken/datadir/spark/tmp

Excessive Worker Disk Usage

Worker work directories accumulate driver‑uploaded files. Schedule regular cleanup of these directories.

Dependency Issues When Submitting Spark Applications

Use the --driver-class-path option to specify required JARs, separating multiple JARs with a colon ( :).

Executor Lost / Shuffle Fetch Failures

Typical causes are insufficient executor memory or too few shuffle partitions. Remedies include increasing executor memory, raising the number of partitions, and adjusting spark.sql.shuffle.partitions for better parallelism.