Spark executor and Databricks cluster calculator
Say how much data the job reads and shuffles. On Databricks, get the worker type and the number of workers to pick; on nodes you already have, see how many executors to run, with how many cores and how much memory. Then see how Spark splits the data into tasks, and copy the settings.
The Spark sizing calculator
Hadoop clusters, Amazon EMR and Google Dataproc run Spark on YARN.
64 GB a node, as on general purpose instances.
96 cores and 384 GB in all
As stored: compressed Parquet or Delta files.
Shuffle Write of the stage in the Spark UI. None for a job with no join, group by or sort.
The answer
Run 17 executors of 5 cores and 19 GB each, 3 on a node: 85 tasks at once. The job reads in 8,192 tasks of 128 MB and its shuffle needs 1,615 partitions.
Each node
- Execution and cache 11.2 GB
- User memory 7.5 GB
- Spark's own 300 MB
- Overhead 1.9 GB
- Executors
- 17
- 3 a node, less 1 for the driver
- Cores each
- 5
- spark.executor.cores
- Tasks at once
- 85
- executors × cores
- Heap
- 19 GB
- spark.executor.memory=19g
- Overhead
- 1.9 GB
- 10% of the heap, at least 384 MB
- Memory a task
- 2.2 GB
- execution and cache ÷ cores, when nothing is cached
Partitions
- Read tasks
- 8,192
- pieces of 128 MB, 97 rounds
- Shuffle partitions
- 1,615
- 127 MB each, 19 rounds
- At the default 200
- 1 GB
- each partition
Watch for
- Left at the default of 200, each shuffle partition would hold 1 GB, above the 256 MB at which adaptive query execution calls a partition skewed. Partitions that large spill to disk or run out of memory, and adaptive query execution merges partitions but splits them only in skewed joins.
- On YARN, Python workers share the 1.9 GB of overhead unless spark.executor.pyspark.memory is set. For Python UDFs or pandas, set it, or raise spark.executor.memoryOverhead, or YARN may stop the container.
The spark-submit line
spark-submit \
--master yarn \
--deploy-mode cluster \
--conf spark.executor.instances=17 \
--conf spark.executor.cores=5 \
--conf spark.executor.memory=19g \
--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.shuffle.partitions=1615 \
your_job.pySpark adds the overhead itself. Check a real run in the Spark UI: spill to disk or long garbage collection is what to tune next.
How the calculator works it out
Each node first keeps 1 core and 1 GB for the operating system and the cluster's own services. The cores left are split into executors of 5 cores at most, because Cloudera found that more than 5 tasks in 1 executor stop getting full write speed from HDFS. Where 5 would leave cores unused, it takes 4 or 3. The room of 1 executor is left for the driver.
Each executor's share of the node's memory holds its heap (spark.executor.memory) and the memory outside it, the overhead. Spark adds 10 percent of the heap as overhead, at least 384 MB, and 40 percent for Python jobs on Kubernetes. Inside the heap, Spark keeps 300 MB for itself and gives 60 percent of the rest to running tasks and cached data together.
Spark reads files in pieces of up to 128 MB, and smaller when the data is too small to give every core a piece. For a shuffle, the calculator sets at least 2 partitions a core, as Spark's tuning guide advises, and none larger than 128 MB, then rounds up so the last round of tasks keeps every core busy.
On Databricks there is 1 executor on each worker with all of its cores. Databricks gives its container 97 percent of the worker's memory less 4,800 MB for its own services, and the heap 80 percent of that container.
To pick the workers, the calculator takes the node family Databricks advises for the kind of job: compute optimized for simple ETL, memory optimized for joins and aggregations, local SSDs for analysis and machine learning. Databricks names total executor memory as what decides how much data stays in memory before spilling to disk, so for each size of that family the calculator counts the workers whose execution memory together is at least what the biggest shuffle writes, and the workers that read the data in no more than the rounds of tasks you allow, and keeps the larger. Rows take more room in memory than in compressed shuffle files, so treat that memory as the least to give the job. The best fit needs the fewest vCPUs, or, where Databricks advises fewer and larger nodes, the largest size that keeps 2 workers for at most a quarter more.
The figures are a place to start: the Spark UI of a real run shows whether tasks spill to disk or wait, and that is what to tune against.
Questions
- How many executors do I need in Spark?
- Take each node's cores less 1, divide them into executors of up to 5 cores, multiply by the nodes and leave 1 executor's room for the driver. On 6 nodes of 16 cores and 64 GB that gives 17 executors of 5 cores, 3 on each node, which run 85 tasks at once.
- How do I calculate spark.executor.memory?
- Divide each node's memory, less 1 GB, by the executors on it, then take off the overhead Spark adds outside the heap: 10 percent, at least 384 MB. On nodes of 64 GB with 3 executors each, that leaves spark.executor.memory at 19g (19 GB), with 1.9 GB of overhead on top.
- What is spark.executor.memoryOverhead?
- It is the memory an executor's container gets beyond its Java heap, for the JVM itself, native libraries and, unless spark.executor.pyspark.memory is set, the Python workers. Spark sets it to 10 percent of spark.executor.memory with a floor of 384 MB, and to 40 percent for non-JVM jobs, such as PySpark, on Kubernetes.
- What should spark.sql.shuffle.partitions be?
- Set it so each partition holds no more than about 128 MB and every core gets at least 2 tasks. For 200 GB shuffled on 85 cores that is 1,615 partitions of about 127 MB each; the default of 200 would make each 1 GB. Adaptive query execution can merge partitions that turn out small, so a number on the high side costs little.
- Why 5 cores per executor?
- Cloudera's tuning guide found that the HDFS client copes badly with many threads, and that at most 5 tasks per executor reach full write speed. Executors of 1 core waste memory on many copies of the JVM and of broadcast data, and very large executors spend long in garbage collection. 5 is a rule of thumb, not a limit.
- How many partitions does Spark read a file into?
- Spark cuts files into pieces of up to 128 MB (spark.sql.files.maxPartitionBytes). When the data spread over the cores comes to less than that, the pieces shrink to give each core one, down to 4 MB. 1 TB read on 85 cores becomes 8,192 tasks, run in 97 rounds.
- Does adaptive query execution set the shuffle partitions for me?
- Partly. Adaptive query execution, on by default since Spark 3.2, merges small shuffle partitions after a stage runs and splits skewed ones in joins, but it starts from spark.sql.shuffle.partitions and cannot split an ordinary partition that is too big. On Databricks, setting spark.sql.shuffle.partitions to auto lets it choose the starting number too.
- How many workers does my Databricks cluster need?
- Enough workers that their execution memory together is at least what the biggest shuffle writes, and that the read takes no more rounds of tasks than you accept. For 1 TB read and 200 GB shuffled on AWS memory optimized workers, in at most 100 rounds, that is 2 workers of r6id.12xlarge, each with 48 vCPUs and 384 GB, and 353 GB of execution memory between them. On all-purpose compute that autoscales, keep the minimum at 2 or more to keep that memory.
- Which worker type should I choose on Databricks?
- Choose memory optimized workers for joins and aggregations, compute optimized workers for simple ETL, and workers with local SSDs for analysis and machine learning. Databricks' compute best practices advise fewer, larger nodes for jobs that shuffle a lot, more memory if tasks spill or run out of memory, less memory for ETL with no join or aggregation, and local SSDs, which Databricks uses for its disk cache, for work that reads the same data again. Types with a d in the name on AWS, such as r6id, carry their own NVMe SSD for shuffle files. Azure Databricks prices no compute optimized workers, so simple ETL there runs on general purpose ones.
- How much memory does a Databricks executor get?
- Databricks runs 1 executor on each worker, with all of its cores. Its container gets 97 percent of the worker's memory less 4,800 MB for Databricks' own services, and spark.executor.memory is 80 percent of that, for most node types: 18,409 MB on a 28 GB node, and 294 GB on r6id.12xlarge. With Photon on, most of it moves off the heap to Photon's engine, so the heap the Spark UI shows is smaller.
Put this calculator on your own site
Copy the code into any web page: a blog post, a course page or your team's wiki. The calculator runs in a frame and shows the same figures as this page, so when the prices change here they change on your page too. It is free and needs no account, and nothing a reader types is sent anywhere.
The code opens the calculator with the settings you have chosen above.
The short script at the end lets the frame grow and shrink with the calculator. Where a page cannot take scripts, choose a fixed height and the frame scrolls.
<iframe src="https://tecyfy.com/embed/spark-calculator" title="Spark sizing calculator" width="100%" height="2950" style="border:0;display:block" loading="lazy" allow="clipboard-write"></iframe>
<p><a href="https://tecyfy.com/data-engineering/spark-calculator">Spark sizing calculator</a> by Tecyfy</p>
<script>addEventListener("message",function(e){if(e.origin!=="https://tecyfy.com"||!e.data||e.data.type!=="tecyfy:height")return;var h=Number(e.data.height);if(!(h>0))return;document.querySelectorAll("iframe").forEach(function(f){if(f.contentWindow===e.source)f.style.height=Math.min(h,20000)+"px"})})</script>Sources
Where each fact comes from
- 1Apache Spark: configuration (executor memory, overhead, cores, memory fraction)
- 2Apache Spark: tuning guide (2 to 3 tasks per core, memory management)
- 3Apache Spark: SQL performance tuning (file partitions, shuffle partitions, adaptive query execution)
- 4Apache Spark source: how a file is split into partitions (FilePartition.maxSplitBytes)
- 5Cloudera: tuning resource allocation (5 cores per executor, heap size and garbage collection)
- 6Cloudera: resource tuning example (6 nodes of 16 cores and 64 GB)
- 7Databricks: adaptive query execution and auto-optimized shuffle
- 8Databricks: 1 executor per worker node
- 9Databricks: compute configuration best practices (node family and size for each kind of job)
- 10Databricks: compute configuration reference (worker and driver types, autoscaling)
- 11Databricks on Google Cloud: compute configuration (local SSDs of 375 GB on n1, n2 and n2d)
- 12Databricks Community, answer by a Databricks employee: container and spark.executor.memory sums
- 13Databricks Community: 18,409 MB of executor memory on a 28 GB node, and less with Photon
- 14AWS: M6i and M6id instances
- 15AWS: R6i and R6id instances
- 16AWS: C6i and C6id instances
- 17AWS: I4i instances
- 18Azure: Ddsv5 sizes
- 19Azure: Edsv5 sizes
- 20Azure Databricks pricing (the worker types Microsoft prices, with no compute optimized series)
- 21Azure: Lsv3 sizes
- 22Google Cloud: N2 machine types
- 23AWS: EC2 instance types (memory per vCPU of compute, general and memory optimized instances)