Scenario: Memory Intensive Join

Scenario: Resolving Executor Out-Of-Memory During Skewed Join

The Challenge

A data engineer is running a Spark SQL query that joins a giant transactional table (1 TB) with a medium-sized advertiser metadata table (5 GB). The join key is heavily skewed (e.g., a few advertising IDs account for 70% of the transactions). When submitting the job with standard parameters, Spark tasks fail repeatedly on worker nodes with YARN errors like: ExecutorLostFailure (executor 4 exited caused by JVM parameter limits / Container killed by YARN for exceeding memory limits).

1. Optimal Spark-Submit Configuration

spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --name "Skewed-Advertiser-Join-ETL" \
    --num-executors 15 \
    --executor-cores 4 \
    --executor-memory 24g \
    --driver-memory 8g \
    --conf spark.yarn.executor.memoryOverhead=6144m \
    --conf spark.sql.adaptive.enabled=true \
    --conf spark.sql.adaptive.skewJoin.enabled=true \
    --conf spark.sql.autoBroadcastJoinThreshold=104857600 \
    --conf spark.memory.fraction=0.8 \
    hdfs:///scripts/skewed_join_job.py

2. Parameter Explanations & Rationale

• --executor-memory 24g / --conf spark.yarn.executor.memoryOverhead=6144m: We significantly increase executor memory and increase the YARN container memory overhead buffer to 6 GB (25% of executor memory instead of the default 10%). This extra buffer prevents YARN from killing containers due to memory overhead spikes during heavy data shuffles.
• spark.sql.adaptive.enabled=true & skewJoin.enabled=true: Activates Adaptive Query Execution (AQE). When Spark detects a skewed partition during execution, it automatically splits the skewed partition into smaller sub-partitions and joins them independently, preventing a single task from running out of memory.
• spark.sql.autoBroadcastJoinThreshold=104857600: Sets the broadcast limit to 100 MB. Since our metadata table is 5 GB, we do not want to force a broadcast join (which would copy 5 GB to all executors and cause Driver/Executor OOMs). This enforces a safe Shuffle Hash Join or Sort Merge Join.
• spark.memory.fraction=0.8: Increases the memory fraction allocated to Spark execution and storage (from the default 0.6 to 0.8) to maximize available heap space for caching skewed partitions during joins.
• Reduced Cores (--executor-cores 4): Allocating 4 cores instead of 5 gives each active thread more available memory (6 GB per core instead of 4.8 GB per core), helping tasks process larger data partitions without spilling to disk or crashing the JVM.