I am using spark on YARN.
Setup:
spark.sql.partitions=3200- 20 executors
- 12Gb+1Gb(memOverhead)/5CPU per executor (~250Gb RAM total)
Simple DAG representation:
- Read data (partition filter + some filters) - partial dropDuplicates() - exchange
- exchange - dropDuplicates() - explode, pythonUDF
- saveAsTable
Some statistics:
- Input ~250Gb, shuffle write ~ 250Gb
- Shuffle read ~250Gb (80Mb per task) (so in deserialized form it takes more than that)
Data after exhange is evenly distrubuted (no skewed keys), explode is small (only 1% of data is added, maximum explosion is 200 records), pythonUDF doesn't create complex structures, just url pasrsing.
Stage 2 is failing with YARN killed executor for running beyond memory limits.
If i ask for twice more executors it works fine. But that's too expensive for cluster.
The question is "why it does not spill computed data to disk and fails with OOM?" It is kind of sad, that spark can't process stage with shuffle read larger than RAM, because everything i've read says, that's not the problem and spark can spill to disk and degrade to mapreduce.