HintsToday

repartition() and coalesce() can help resolve OOM (Out of Memory) errors in Spark — but how you use them depends on the root cause of the memory pressure.

Let’s break it down clearly.

🧠 First: What causes OOM in Spark?

OOM (OutOfMemoryError) in Spark usually happens when:

• A task processes too much data (input partition is too large)
• There’s data skew: one partition has most of the data
• Too many small tasks and high task scheduling overhead
• A shuffle stage pulls in too much intermediate data
• Caching large datasets without enough executor memory

✅ How repartition() and coalesce() help

✅ 1. repartition() helps avoid OOM due to large partitions

If you’re doing something like:

large_df = spark.read.parquet("huge_data")
result = large_df.groupBy("country").agg(...)

Symptoms of OOM:

• Only a few tasks doing all the work
• Spark UI shows huge partition sizes (like 500MB, 1GB)

✅ Fix:

# Break data into more partitions before a wide operation
large_df = large_df.repartition(200)  # or based on a column

This spreads the data across more tasks, reducing per-task memory pressure.

✅ 2. coalesce() helps avoid OOM from too many tasks

If your data has many small files (common in S3/HDFS) and you do:

df = spark.read.json("s3://many-small-files/")
df.count()

You may end up with 1000s of partitions, causing:

• High scheduling overhead
• Memory issues due to many simultaneous tasks

✅ Fix:

df = df.coalesce(50)  # Reduce to fewer partitions without shuffle

This makes the job more manageable and reduces memory strain from task overhead.

⚠️ Important Tips

🔧 Example: Fixing OOM in Practice

# You're reading a 50GB dataset
df = spark.read.parquet("/data/big")

# BEFORE: Only 8 partitions (default), each 6+ GB — causes OOM
df.rdd.getNumPartitions()
# >>> 8

# FIX: Increase partitions before aggregation
df = df.repartition(100)  # Now each partition ~500MB

# Proceed with join or groupBy safely
result = df.groupBy("user_id").agg(...)

🔄 Combine both for optimal memory control

df = df.repartition(200)  # For better memory use during shuffle
... # run heavy operations
df = df.coalesce(50)      # Reduce output files during write
df.write.parquet("final_output/")

✅ Conclusion

Great follow-up! Let’s compare how data is distributed in:

• repartition(100)
• repartition('col', 40)
• coalesce()
• bucketing

Each has very different distribution logic — here’s a clear and complete explanation 👇

🔁 1. repartition(100): Random & Even Distribution (with shuffle)

• ✅ Shuffles data randomly across 100 partitions.
• 🔁 Spark reshuffles all rows with a round-robin-like strategy.
• ✅ This gives even-sized partitions, assuming no data skew.

df = df.repartition(100)

📌 Use when:

• You want to increase partitions
• Fix large partition size causing OOM
• Preparing for wide transformations (join, groupBy)

💡 Good for memory load balancing across executors.

🎯 2. repartition('col', 40): Hash Partitioning by Column (with shuffle)

• ✅ Spark hashes the col values and assigns rows to 40 partitions.
• ✅ Ensures all rows with same key (col) go to the same partition.

df = df.repartition(40, "country")

📌 Use when:

• You’re preparing for join or groupBy on col
• You want co-location of keys for better performance

⚠️ Even distribution only if the values in col are not skewed.

• If one value appears much more than others → you’ll get data skew.

⚠️ 3. coalesce(20): Skewed, No Shuffle, Merge-only

• ❌ No shuffle.
• 🔁 It combines existing partitions by reducing them.
• Often results in uneven partition sizes.

df = df.coalesce(20)

📌 Use when:

• You’re reducing number of partitions after filtering
• Before writing to avoid too many small output files

⚠️ Does NOT distribute evenly. Example:

• If you had 100 partitions with 1MB each
• Coalescing to 5: first few partitions may be 10–20MB, last few 1MB

📦 4. Bucketing (bucketBy): Even by Hash, but only at Write-Time

df.write.bucketBy(8, "id").sortBy("id").saveAsTable("users")

• ✅ Spark hashes on id, distributes data evenly across 8 buckets
• 🔁 Applied at write time only (needs saveAsTable)
• Later joins on bucketed column are more efficient

📌 Use when:

• You write Delta or Hive tables
• Want to optimize read performance (especially for joins)

⚠️ You need to:

• Save as table
• Enable bucketing
• Sometimes manually read with spark.sql("SELECT ...")

🔍 Summary Table

🔧 Rule of Thumb

Here’s a short and clear solution for each common OOM cause in Spark:

✅ 1. Large Partition → Task processes too much data

• Fix: Use repartition(n) to increase number of partitions and reduce per-task data size.

✅ 2. Data Skew → One partition has most of the data

• Fix: Use salting/skew hint or repartition by key with pre-processing.
• Ex: Add random prefix to skewed key before join.

✅ 3. Too Many Small Tasks → High overhead

• Fix: Use coalesce(n) to reduce number of small partitions.

✅ 4. Shuffle Overload → Too much intermediate data in memory

• Fix:
  • Use broadcast join for small tables
  • Tune spark.sql.shuffle.partitions
  • Increase spark.memory.fraction if needed

✅ 5. Caching Large Dataset → Insufficient memory

• Fix:
  • Use persist(StorageLevel.DISK_ONLY) or selective caching
  • Unpersist unused RDDs/DataFrames with .unpersist()
  • Avoid caching large intermediate joins

Here’s your Spark Configuration Tuning Cheat Sheet 🧠⚙️
Organized by problem area, with optimal settings, defaults, and tips for solving OOM errors, slow jobs, skew, and shuffle issues.

🚀 GENERAL PERFORMANCE

💥 OOM ERROR RESOLUTION

📦 MEMORY & STORAGE

🔁 SHUFFLE OPTIMIZATION

📊 JOIN OPTIMIZATION

🔥 SKEW HANDLING

🧪 DEBUGGING & DIAGNOSTICS

⚠️ Bonus: Safe Memory Limits

Pages: 1 2 3 4