Spark Performance Tuning I : Setting appropriate number of Executors :

Executor :

An executor is a single JVM process which is launched for an application on a worker node.

Executor runs tasks and keeps data in memory or disk storage across them. Each application has its own executors. A single node can run multiple executors depending on the number of cores, if a node has n cores then a maximum of n executors can run on that node.

An executor stays up for the duration of the Spark Application and runs the tasks in multiple threads. The number of executors for a spark application can be specified inside the SparkConf or via the flag –num-executors from command-line.

Cores : A core is a basic computation unit of CPU and a CPU may have one or more cores to perform tasks at a given time. The more cores we have, the more work we can do. In spark, this controls the number of parallel tasks an executor can run.

There are two ways in which we can configure number of executor and core to the Spark job:

Static Allocation

Dynamic Allocation

Static Allocation : Here we pass number of executors explicitly via command line or SparkConf.

Lets Suppose below scenario where we will be calculating number of cores/executors/memory needed :

6 Nodes and each node have 16 cores, 64 GB RAM

First on each node, 1 core and 1 GB is needed for OS and Hadoop Daemons, so we have 15 cores, 63 GB RAM for each node.

No of cores : So It might appears, more parallel tasks for each executor will give better performance.

But research shows that any application with more than 5 concurrent tasks, would lead to a bad show.

So the optimal value is 5.

Even if we double the available cores per node as 32 then also 5 core per executor will be the optimal number.

No of executors : As we have 5 cores each executor and total of 15 cores each node. So number of executors per node will come

out to be 15/5=3.

So with 6 nodes = 3x6 = 18 total executors for whole application

But 1 executor is needed for Application Master

so 18-1 = 17 Executors

Memory per executor : Total executor per node is 3 and memory for each node is 63 GB.

So memory per executor = 63/3 = 21 GB

But here we need to calculate memory overhead for each executor, formula for calculating the same is :

max(384 MB , .07 * spark.executor.memory)

Here, max(384 MB , .07 * 21 GB)

= max(384 MB , 1.47 GB)

=1.47 GB

So overhead memory needed is 1.47 GB

So subtarcting this overhead memory from total memory to get estimate of memory per executor :

=(21-1.47)GB

=19 GB Approx

So conclusion: 5 Cores per executor, 3 executors per node and 19 GB per executor is needed.

Exercise : Calculate the same for 6 Nodes and Each node have 32 Cores, 64 GB

Final numbers : Executors – 35, Cores 5, Executor Memory – 9 GB

Dynamic Allocation : The values are picked up based on the requirement (size of data, amount of computations needed) and released after use. This helps the resources to be re-used for other applications.

To enable dynamic allocation we need to set this property as True :

spark.dynamicAllocation.enabled

After setting this property as true we need not to specify number of executors because

the static parameter numbers we give at spark-submit is for the entire job duration.

However if dynamic allocation comes into picture, there would be different stages like the following:

Initial Number of Executors :

Property - spark.dynamicAllocation.initialExecutors

Initial number of executors to run if dynamic allocation is enabled.

If `--num-executors` (or `spark.executor.instances`) is set and larger than this value,

it will be used as the initial number of executors.

Min and max limit on number of executors :

Property - spark.dynamicAllocation.minExecutors :

Lower bound for the number of executors if dynamic allocation is enabled.

Default value : 0

Also, this is the default value of spark.dynamicAllocation.initialExecutors property as well.

Property - spark.dynamicAllocation.maxExecutors :

Upper bound for the number of executors if dynamic allocation is enabled.

Default value is Infinity.So this says that spark application can eat away all the resources if needed.

Asking for new executors :

Property - spark.dynamicAllocation.schedulerBacklogTimeout :

If dynamic allocation is enabled and there have been pending tasks backlogged for more than this duration, new executors will be requested.

Giving up current executors :

Property - spark.dynamicAllocation.executorIdleTimeout :

If dynamic allocation is enabled and an executor has been idle for more than this duration, the executor will be removed.

Keep Learning keep growing !!