Why OOM: Spark attempted a shuffle/join where one side (right) was assumed small and likely broadcastable. At 100M rows it can't be broadcast to executors; if Spark still tries (or the join causes huge shuffle/hash tables), executors build large in-memory hash maps and spill/OOM. Also collecting the final result brings all data to driver, risking driver OOM.Safer approach — principles:- Don’t broadcast unless you’ve validated size.- Push aggregation/predicate down before the join to reduce data.- Repartition both sides by join key to make the join shuffle-local and enable map-side combining.- Use map-side reduction (combineByKey / reduceByKey) or DataFrame aggregations after repartition so less data is shuffled.- Detect and handle skew via salting on hot keys.- Don’t use collect(); write results to durable storage or limit transfer.Example implementation (PySpark):python
from pyspark.sql import functions as F
left = spark.read.parquet('s3://data/large_left')
right = spark.read.parquet('s3://data/small_right')
# 1) If possible, pre-aggregate right to smallest form (push-down)
right_agg = right.groupBy('user_id').agg(F.sum('whatever_col').alias('r_val'))
# 2) Estimate size; only broadcast if small
approx_size_bytes = spark._jsparkSession.sessionState().executePlan(right_agg._jdf.queryExecution().logical()).optimizedPlan().stats().sizeInBytes()
BROADCAST_THRESHOLD = 50 * 1024 * 1024 # e.g. 50MB
if approx_size_bytes < BROADCAST_THRESHOLD:
joined = left.join(F.broadcast(right_agg), on='user_id')
result = joined.groupBy('country').agg(F.sum('amount').alias('total_amount'))
result.write.mode('overwrite').parquet('s3://out/agg_by_country')
else:
# 3) Repartition both sides by join key to avoid building huge broadcast hashmaps
num_parts = 200 # tune to cluster
left_r = left.repartition(num_parts, 'user_id')
right_r = right_agg.repartition(num_parts, 'user_id')
# 4) Perform join and map-side combine via aggregation after repartition
joined = left_r.join(right_r, on='user_id', how='inner')
# 5) If skew suspected, salt hot keys:
# Add salt column to both sides for known hot user_ids, join on (user_id, salt)
# omitted for brevity; implement by exploding salt range for hot keys
agg = joined.groupBy('country').agg(F.sum('amount').alias('total_amount'))
# 6) Persist or write out instead of collect
agg.write.mode('overwrite').parquet('s3://out/agg_by_country')
python
pairs = joined.select('country', 'amount').rdd.map(lambda r: (r['country'], r['amount']))
reduced = pairs.reduceByKey(lambda a, b: a + b)
reduced.toDF(['country','total_amount']).write.parquet('s3://out/agg_by_country')
