Spark cluster: Launched executors less than specified

Hi @RonyA 

You haven't shared what is your dataset size. Apart from data you need to tune on few spark parameters.

spark = (SparkSession
    .builder.master("yarn")
    .config("spark.executor.cores", "5") # you have mentioned 12
    .config("spark.num.executors", "10")
    .config("spark.executor.memory", "10G")
    .config("spark.executor.memoryOverhead", "2G") # executor memory * 0.1 or 0.2 % 
    .config("spark.driver.memory", "10G")
    .config("spark.driver.memoryOverhead", "2G")   # driver memory * 0.1 or 0.2 % 
    .config("spark.sql.hive.convertMetastoreOrc", "true")
    .config("spark.executor.heartbeatInterval", "60s") # default 10s
    .config("spark.network.timeout", "600s")           # default 120s
    .config("spark.driver.maxResultSize", "2g")
    .config("spark.driver.cores","4")
    .config("spark.executor.extraClassPath", "-Dhdp.version=current")
    .config("spark.debug.maxToStringFields", 200)
    .config("spark.sql.catalogImplementation", "hive")
    .config("spark.memory.fraction", "0.8")
    .config("spark.memory.storageFraction", "0.2")
    .config("spark.sql.hive.filesourcePartitionFileCacheSize", "0")
    .config("spark.yarn.maxAppAttempts", "10")
    .appName(app_name)
    .enableHiveSupport().getOrCreate())

Apart from above if you are doing any kind of wide operation shuffle is involved. To set shuffle value we will use below calculation:

spark.sql.shuffle.partitions = shuffle input size/ hdfs block size 

for example, shuffle input size is 10GB and hdfs block size is 128 MB then shuffle partitions is 10GB/128MB = 80 partitions.

And also check you have enabled dynamic allocation or not. You can open Spark UI --> Select Application --> Go to the Environment page --> find spark.dynamicallocation.enabled property.

I see 2 things that would be good to understand:

1)why do the yarn containers exceed their size

2)why does he not provide the number of executors that you request?

1)

It seems like you are exceeding the yarn container size of 10GB. The executors will run in yarn containers.   Maybe you need to increase the minimum yarn container size a bit? I think the message suggests the minimum container size for yarn is 10GB. If you request a 8GB executor, and there is some (2GB)overhead, he might hit the ceiling of what was assigned to him and this executor will exit.

2)

Looks like your cluster is not capable of providing the requested 10 executors of 8GB?

Other relevant info to share would be: how many nodes do you have, and for each node, how much memory is assigned to yarn , and how much is the yarn minimum container size?
Example:

suppose the yarn container size is 3 GB. Suppose you have 9 nodes. Suppose your executor memory is 1GB . Suppose 10 GB on each node is allocated to yarn. This means you have on each node enough memory to start 3 containers (3x3GB< 10GB). THerefore, when dynamic allocation is enabled, he will start 27 execuctors. Even if you would ask for more than this 27, he will only be capable of providing 27.

Maybe this helps?