Error: Executor Memory Issue with Broadcast Joins in Structured Streaming – Unable to Store 69–80 MB

[pooja_bhumandla]

Hi Community,

I encountered the following error:

      Failed to store executor broadcast spark_join_relation_1622863 (size = Some(67141632)) in BlockManager              with storageLevel=StorageLevel(memory, deserialized, 1 replicas)

in a Structured Streaming job in Databricks with foreachBatch writing to a Delta table. 

I’ve observed that most of the failures occurred when table sizes were in the range of 69–75 MB, and the error suggests that Spark is unable to store the broadcasted table in memory.

When reviewing executors memory usage,

I noticed there was a few GBs of free memory available, but there was also high swap usage.
Given the free memory available, I would expect the executor to be able to hold the 69–80 MB table for broadcasting.

• Why couldn’t it hold this data around 80MB despite having free memory in GBs?
• Even if I disable the broadcast setting, I believe MERGE operations still enforce broadcasting internally.
• Is this error primarily due to the broadcast threshold, or is it related to insufficient memory in the executor?
• Since the error occurs when the executor cannot hold around 69–80 MB in memory, to handle this - should I increase the broadcast threshold to 100MB or decrease it? 

Looking forward to hearing your thoughts and suggestions to solve this error!

[ManojkMohan]

@pooja_bhumandla 

1. Join strategy

• For the Delta MERGE, try to ensure the large side is not broadcast by setting autoBroadcastJoinThreshold low or disabling it https://kb.databricks.com/sql/bchashjoin-exceeds-bcjointhreshold-oom

• If you are explicitly broadcasting a reference DataFrame, remove the hint or replace with a shuffle join hint

2 Memory and cluster configuration

• Increase executor memory and memory overhead if the job is genuinely heavy
• Reduce the number of cores per executor to give each task more memory headroom

3 Micro‑batch load

If the size of the source side of the MERGE grows over time, consider limiting micro‑batch size

https://docs.databricks.com/aws/en/structured-streaming/foreach

If you share your cluster specs (executor memory/cores, threshold settings, and rough sizes of the tables on each side of the MERGE), some more solutioning brainstorming can be done

[Yogesh_Verma_]

What Spark Does During a Broadcast Join-

• Spark identifies the smaller table (say 80MB).
• The driver collects this small table to a single JVM.
• The driver serializes the table into a broadcast variable.
• The broadcast variable is shipped to all executors.
• Executors store it inside the BlockManager storage region.
• Each executor loads it into memory to build a hash map for fast joining.