-2

I am running spark job with cluster mode in EMR 5.27.0. EMR comes with dynamic spark allocation property set to true.

Now when i start spark job or even start spark shell i can see many executors launched in Spark UI.

Why this is happening even though i am using just spark-shell?

I tried multiple things like setting properties like spark.dynamicAllocation.initialExecutors=1 but no success.

Ideally dynamic allocation should allocate initialExecutors first then launch more after spark.dynamicAllocation.schedulerBacklogTimeout and there are waiting/pending tasks properties were satisfied. But its initially starting all executors at once.

Whats role of spark.dynamicAllocation.executorAllocationRatio this property i tried controlling executors by using this property but it didn't work.

When i start EMR Shell with minExecutor=1 then i am getting following log INFO: Utils: spark.executor.instances less than spark.dynamicAllocation.minExecutors is invalid, ignoring its setting, please update your configs.

In logs its showing spark.executor.instances = 50.

I verified spark-default but it has no such properties.

Please help me to understand this behaviour.

1

1 Answer 1

3

When Spark reads a file from HDFS, it creates a single partition for a single input split. Input split is set by the Hadoop InputFormat used to read this file. For instance, if you use textFile() it would be TextInputFormat in Hadoop, which would return you a single partition for a single block of HDFS (but the split between partitions would be done on line split, not the exact block split), unless you have a compressed text file. In case of compressed files depending on the type of compression the number of partitions vary

There are few other parameters mentioned below that typically work with RDD's and not Dataframes.

spark.default.parallelism - For distributed shuffle operations like reduceByKey and join, the largest number of partitions in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster manager:
Local mode: number of cores on the local machine
Mesos fine grained mode: 8
Others: total number of cores on all executor nodes or 2, whichever is larger

and

spark.sql.shuffle.partitions - controls the number of partitions for operations on DataFrames (default is 200)

Once you have the number of partitions defined, each partitioned is processed by a task and each task runs on an executor instance. With Dynamic allocation the number of executor instances are controlled by the number of partitions, which can change at every stage of DAG execution.

If you want to control the number of executors when Dynamic Allocation is turned on then you can set the below configurations in spark default configuration file.

spark.dynamicAllocation.initialExecutors | spark.dynamicAllocation.minExecutors |   Initial number of executors to run if dynamic allocation is enabled.
spark.dynamicAllocation.maxExecutors     | infinity                              | Upper bound for the number of executors if dynamic allocation is enabled.
spark.dynamicAllocation.minExecutors     | 0                                     | Lower bound for the number of executors if dynamic allocation is enabled.

You should set spark.dynamicAllocation.maxExecutors to control the maximum number of executors that can be provisioned in the EMR cluster.

The ExecutorAllocationManager has a sophisticated algorithm to determine the number of executors that are spawned:-

private def updateAndSyncNumExecutorsTarget(now: Long): Int = synchronized {
    val maxNeeded = maxNumExecutorsNeeded

    if (initializing) {
      // Do not change our target while we are still initializing,
      // Otherwise the first job may have to ramp up unnecessarily
      0
    } else if (maxNeeded < numExecutorsTarget) {
      // The target number exceeds the number we actually need, so stop adding new
      // executors and inform the cluster manager to cancel the extra pending requests
      val oldNumExecutorsTarget = numExecutorsTarget
      numExecutorsTarget = math.max(maxNeeded, minNumExecutors)
      numExecutorsToAdd = 1

      // If the new target has not changed, avoid sending a message to the cluster manager
      if (numExecutorsTarget < oldNumExecutorsTarget) {
        client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks, hostToLocalTaskCount)
        logDebug(s"Lowering target number of executors to $numExecutorsTarget (previously " +
          s"$oldNumExecutorsTarget) because not all requested executors are actually needed")
      }
      numExecutorsTarget - oldNumExecutorsTarget
    } else if (addTime != NOT_SET && now >= addTime) {
      val delta = addExecutors(maxNeeded)
      logDebug(s"Starting timer to add more executors (to " +
        s"expire in $sustainedSchedulerBacklogTimeoutS seconds)")
      addTime += sustainedSchedulerBacklogTimeoutS * 1000
      delta
    } else {
      0
    }
  }

In the above code snippet the high level flow is as follows:-

max_needed = (Sum of Running + Pending Tasks)/Tasks per executor

If this value is less than the numExecutorsTarget which is max of initial executors and spark.executor.instancesthen the initial request for adding executors is determined based onspark.dynamicAllocation.schedulerBacklogTimeoutand every subsequent request is made on the duration defined byspark.dynamicAllocation.sustainedSchedulerBacklogTimeout` and executors are spun up in multiples of 2 opposite of exponential backoffs.

Finally, an important point to note is the variable addTime is set during the submission of stage, so whatever time you have set is not from when the job is running but from when the stage was submitted.

For the default configuration of the EMR cluster you can refer to the documentation present here - https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-configure.html

6
  • Hi Jayadeep thanks for detailed explanation. I have just edited my question. Referring here again that ideally dynamic allocation should allocate initialExecutors first then launch more after spark.dynamicAllocation.schedulerBacklogTimeout and there are waiting/pending tasks properties are satisfied. But its initially starting all executors at once. Also i tried controlling it using spark.dynamicAllocation.executorAllocationRatio. It didn't work. Nov 20, 2019 at 1:07
  • I have updated my answer with further details for your questions. If this was helpful please do accept the answer. Nov 20, 2019 at 3:18
  • HI Jaydeep still i am facing same issue not able to figure it out. This is my log in EMR Shell - INFO: Utils: spark.executor.instances less than spark.dynamicAllocation.minExecutors is invalid, ignoring its setting, please update your configs. I am not able understand why its doing this. Nov 20, 2019 at 6:34
  • I am not sure of how EMR works, you might want to check with AWS support but Spark works the way I mentioned. If this answer was helpful please do accept the answer. Nov 20, 2019 at 9:26
  • You are right about spark i think something to do with EMR. Nov 20, 2019 at 11:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.