# How to Optimize Spark Jobs for Performance and Troubleshoot OutOfMemoryError

> Optimize Spark jobs and fix OutOfMemoryError by tuning memory, managing broadcasts, using bucket joins, and minimizing shuffle data for better performance. Learn expert tips now.

- Repository: [DataExpert.io/data-engineer-handbook](https://github.com/DataExpert-io/data-engineer-handbook)
- Tags: performance
- Published: 2026-08-12

---

**You can eliminate Spark OutOfMemoryError exceptions and improve job performance by tuning heap and off-heap memory settings, disabling automatic broadcasts for large tables, implementing bucket joins on high-cardinality keys like `match_id`, and using `sortWithinPartitions` to minimize shuffle data.**

Apache Spark's distributed architecture requires careful tuning to avoid memory exhaustion and costly shuffles. In the DataExpert-io/data-engineer-handbook repository, the Spark fundamentals boot camp demonstrates production-ready patterns for optimizing Spark jobs and resolving "Java heap space" errors. This guide extracts the exact configuration strategies and join optimizations used in the course materials to help you scale your data pipelines efficiently.

## Configuring Memory Management to Prevent OOM

The [`intermediate-bootcamp/materials/3-spark-fundamentals/README.md`](https://github.com/DataExpert-io/data-engineer-handbook/blob/main/intermediate-bootcamp/materials/3-spark-fundamentals/README.md) file identifies insufficient heap configuration as the primary cause of "Java heap space" OutOfMemoryError exceptions. Proper memory tuning starts with three key areas: executor heap sizing, off-heap allocation, and garbage collection strategy.

### Executor Heap and Off-Heap Settings

Increase driver and executor memory to match your workload characteristics. For production workloads processing large datasets, configure at least `4g` to `8g` per executor:

```python
spark.conf.set("spark.executor.memory", "8g")
spark.conf.set("spark.driver.memory", "4g")

```

Enable off-heap memory to reduce GC pressure for cached datasets:

```python
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "2g")

```

### Garbage Collection Optimization

Use modern GC algorithms to handle large heaps with minimal pause times. Set these JVM options when launching your Spark session:

- **G1GC**: `-XX:+UseG1GC` – suitable for heaps up to 16GB
- **ZGC**: `-XX:+UnlockExperimentalVMOptions -XX:+UseZGC` – ideal for very large heaps requiring minimal latency

## Optimizing Join Strategies for Performance

The homework materials in [`intermediate-bootcamp/materials/3-spark-fundamentals/homework/homework.md`](https://github.com/DataExpert-io/data-engineer-handbook/blob/main/intermediate-bootcamp/materials/3-spark-fundamentals/homework/homework.md) demonstrate specific join patterns for handling dimension tables (`medals`, `maps`) and fact tables (`match_details`, `matches`, `medal_matches_players`).

### Disable Automatic Broadcast for Large Tables

Prevent Spark from automatically broadcasting large tables by setting the threshold to `-1`:

```python
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

```

### Explicitly Broadcast Small Dimension Tables

Broadcast only small lookup tables to avoid shuffle overhead:

```python
spark.conf.set("spark.sql.broadcastTimeout", "1200")
medals = spark.broadcast(medals_df)
maps = spark.broadcast(maps_df)

```

### Implement Bucket Joins for Large Fact Tables

Bucket large fact tables on the join key (`match_id`) with **16 buckets** to co-locate data and eliminate shuffle operations. According to the homework guidelines, this is the optimal strategy for joining `match_details`, `matches`, and `medal_matches_players`:

```python

# Repartition to match bucket count

match_details = match_details.repartition(16, "match_id")
matches = matches.repartition(16, "match_id")
medal_players = medal_matches_players.repartition(16, "match_id")

# Join bucketed tables on match_id

joined = match_details.join(matches, "match_id") \
    .join(medal_players, "match_id")

```

## Reducing Shuffle Through Partitioning

Shuffle operations are the primary performance bottleneck in distributed Spark jobs. The handbook recommends two techniques to minimize data movement: strategic partitioning and sort-based optimization.

### Partition by Low-Cardinality Columns

Use low-cardinality columns (such as `playlist` and `map`) as partition keys to ensure even data distribution across nodes. This prevents skew and reduces the memory footprint per task.

### Apply sortWithinPartitions

The homework explicitly suggests experimenting with `sortWithinPartitions` to achieve the smallest shuffle size. Sort data within partitions after joining to maintain locality:

```python
joined_df = match_details.join(matches, "match_id")
joined_df = joined_df.sortWithinPartitions("playlist", "map")

```

### Caching and Persistence

Cache only necessary intermediate results using `MEMORY_AND_DISK` storage to avoid recomputation while preventing pure memory pressure:

```python
from pyspark import StorageLevel

joined_df.persist(StorageLevel.MEMORY_AND_DISK)

# ... perform aggregations ...

joined_df.unpersist()  # Free heap immediately after use

```

## Troubleshooting OutOfMemoryError

When Spark throws "Java heap space" errors, the [`README.md`](https://github.com/DataExpert-io/data-engineer-handbook/blob/main/README.md) in the Spark fundamentals section points to specific diagnostic steps and fixes.

### Diagnostic Steps

1. **Check Spark UI**: Navigate to the "Storage" and "Executors" tabs to identify memory spill to disk and heap utilization percentages.
2. **Enable Event Logging**: Set `spark.eventLog.enabled=true` and inspect logs for "GC overhead limit exceeded" messages.
3. **Verify Memory Fraction**: Ensure `spark.memory.fraction` is set to approximately `0.6` to allocate sufficient memory for execution versus storage.

### Immediate Fixes

- **Increase driver/executor memory** if UI shows "Heap Memory Used" approaching the configured limit
- **Tune `spark.memory.fraction`** to 0.6 (default is 0.6, but lowering to 0.5 can help storage-heavy workloads)
- **Reduce partition size** by increasing `spark.sql.shuffle.partitions` to 400 or higher for very large datasets

## Production-Ready Optimization Example

Place this complete job in [`src/jobs/optimized_match_analysis.py`](https://github.com/DataExpert-io/data-engineer-handbook/blob/main/src/jobs/optimized_match_analysis.py) (following the repository structure for Spark jobs):

```python
from pyspark.sql import SparkSession
from pyspark import StorageLevel

spark = SparkSession.builder \
    .appName("OptimizedMatchAnalysis") \
    .getOrCreate()

# Memory configuration to prevent OOM

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "2g")
spark.conf.set("spark.executor.memory", "8g")
spark.conf.set("spark.driver.memory", "4g")
spark.conf.set("spark.memory.fraction", "0.6")

# Load datasets

matches = spark.read.parquet("s3://data/matches")
match_details = spark.read.parquet("s3://data/match_details")
medals = spark.read.parquet("s3://data/medals")
maps = spark.read.parquet("s3://data/maps")
medal_players = spark.read.parquet("s3://data/medal_matches_players")

# Broadcast small dimension tables

spark.conf.set("spark.sql.broadcastTimeout", "1200")
medals = spark.broadcast(medals)
maps = spark.broadcast(maps)

# Bucket-join large fact tables on match_id with 16 buckets

match_details = match_details.repartition(16, "match_id")
matches = matches.repartition(16, "match_id")
medal_players = medal_players.repartition(16, "match_id")

joined = match_details \
    .join(matches, "match_id") \
    .join(medal_players, "match_id") \
    .join(medals.value, "medal_id") \
    .join(maps.value, "map_id")

# Optimize shuffle with sortWithinPartitions

joined = joined.sortWithinPartitions("playlist", "map")

# Cache selectively

joined.persist(StorageLevel.MEMORY_AND_DISK)

# Perform aggregations

player_kills = joined.groupBy("player_id") \
    .agg({"kills": "avg"}) \
    .withColumnRenamed("avg(kills)", "avg_kills_per_game")

# Write results

player_kills.write.mode("overwrite").parquet("s3://output/player_kills")

# Cleanup

joined.unpersist()
spark.stop()

```

## Summary

- **Increase heap sizes** using `spark.executor.memory` and `spark.driver.memory` to resolve "Java heap space" errors documented in the Spark fundamentals README.
- **Disable automatic broadcasts** with `spark.sql.autoBroadcastJoinThreshold=-1` and explicitly broadcast only small tables like `medals` and `maps`.
- **Implement 16-bucket joins** on high-cardinality keys like `match_id` to co-locate data and eliminate shuffles between fact tables.
- **Apply `sortWithinPartitions`** after joins to minimize shuffle size, as recommended in the course homework materials.
- **Use off-heap memory** and modern GC algorithms (G1GC or ZGC) to reduce garbage collection pauses on large datasets.

## Frequently Asked Questions

### What causes OutOfMemoryError in Spark?

OutOfMemoryError in Spark typically occurs when the executor or driver heap is insufficient for the working set of data, or when garbage collection cannot reclaim memory fast enough. According to the DataExpert-io/data-engineer-handbook Spark fundamentals guide, the most common symptom is the "Java heap space" error appearing in logs when processing large joins or aggregations. Increasing `spark.executor.memory` and enabling off-heap storage usually resolves this.

### When should I use bucket joins versus broadcast joins?

Use **broadcast joins** for small dimension tables (typically under 10MB) that can fit comfortably in memory across all nodes, such as the `medals` and `maps` tables referenced in the bootcamp homework. Use **bucket joins** for large fact tables (millions of rows) that would cause memory pressure if broadcast. The handbook recommends bucketing tables like `match_details` and `matches` on `match_id` with 16 buckets to enable efficient join execution without shuffling.

### How do I choose the right number of buckets?

The DataExpert-io/data-engineer-handbook specifically recommends **16 buckets** for joining tables on `match_id`, which provides sufficient parallelism for medium-sized clusters while avoiding too many small files. For larger datasets (terabyte scale), increase to 64 or 128 buckets. The key is to ensure the number of buckets matches or is a multiple of your `spark.sql.shuffle.partitions` to maintain data locality during joins.

### What is the optimal memory fraction setting?

Set `spark.memory.fraction` to **0.6** (the default) for balanced workloads, or reduce to **0.5** if you experience OOM errors during heavy caching operations. This parameter controls the ratio between execution memory and storage memory. According to the repository's optimization patterns, keeping 40% of the heap reserved for user data structures and Spark internal metadata prevents heap exhaustion during complex transformations.