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 on
spark.dynamicAllocation.schedulerBacklogTimeoutand every subsequent request is made on the duration defined by
spark.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