[
https://issues.apache.org/jira/browse/SPARK-22958?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Shaoquan Zhang updated SPARK-22958:
-----------------------------------
Description:
We have encountered the following scenario. We run a very simple job in yarn
cluster mode. This job needs only one executor to complete. In the running,
this job was stuck forever.
After checking the job log, we found an issue in the Spark. When executor fails
to register with driver, YarnAllocator is blind to know it. As a result, the
variable (numExecutorsRunning) maintained by YarnAllocator does not reflect the
truth. When this variable is used to allocate resources to the running job,
misunderstanding happens. As for our job, the misunderstanding results in
forever stuck.
The more details are as follows. The following figure shows how executor is
allocated when the job starts to run. Now suppose only one executor is needed.
In the figure, step 1,2,3 show how the executor is allocated. After the
executor is allocated, it needs to register with the driver (step 4) and the
driver responses to it (step 5). After the 5 steps, the executor can be used to
run tasks.
!How new executor is registered.png!
In YarnAllocator, when step 3 is finished, it will increase the the variable
(numExecutorsRunning) as show in the following code.
{code:java}
def updateInternalState(): Unit = synchronized {
numExecutorsRunning += 1
executorIdToContainer(executorId) = container
containerIdToExecutorId(container.getId) = executorId
val containerSet =
allocatedHostToContainersMap.getOrElseUpdate(executorHostname,
new HashSet[ContainerId])
containerSet += containerId
allocatedContainerToHostMap.put(containerId, executorHostname)
}
if (numExecutorsRunning < targetNumExecutors) {
if (launchContainers) {
launcherPool.execute(new Runnable {
override def run(): Unit = {
try {
new ExecutorRunnable(
Some(container),
conf,
sparkConf,
driverUrl,
executorId,
executorHostname,
executorMemory,
executorCores,
appAttemptId.getApplicationId.toString,
securityMgr,
localResources
).run()
updateInternalState()
} catch {
case NonFatal(e) =>
logError(s"Failed to launch executor $executorId on container
$containerId", e)
// Assigned container should be released immediately to avoid
unnecessary resource
// occupation.
amClient.releaseAssignedContainer(containerId)
}
}
})
} else {
// For test only
updateInternalState()
}
} else {
logInfo(("Skip launching executorRunnable as runnning Excecutors count:
%d " +
"reached target Executors count: %d.").format(numExecutorsRunning,
targetNumExecutors))
}
{code}
Imagine the step 4 is failed due to some reason (for example network
fluctuation).
was:
We have encountered the following scenario. We run a very simple job in yarn
cluster mode. This job needs only one executor to complete. In the running,
this job was stuck forever.
After checking the job log, we found an issue in the Spark. When executor fails
to register with driver, YarnAllocator is blind to know it. As a result, the
variable (numExecutorsRunning) maintained by YarnAllocator does not reflect the
truth. When this variable is used to allocate resources to the running job,
misunderstanding happens. As for our job, the misunderstanding results in
forever stuck.
The more details are as follows. The following figure shows how
!How new executor is registered.png!
> Spark is stuck when the only one executor fails to register with driver
> -----------------------------------------------------------------------
>
> Key: SPARK-22958
> URL: https://issues.apache.org/jira/browse/SPARK-22958
> Project: Spark
> Issue Type: Bug
> Components: YARN
> Affects Versions: 2.1.0
> Reporter: Shaoquan Zhang
> Attachments: How new executor is registered.png
>
>
> We have encountered the following scenario. We run a very simple job in yarn
> cluster mode. This job needs only one executor to complete. In the running,
> this job was stuck forever.
> After checking the job log, we found an issue in the Spark. When executor
> fails to register with driver, YarnAllocator is blind to know it. As a
> result, the variable (numExecutorsRunning) maintained by YarnAllocator does
> not reflect the truth. When this variable is used to allocate resources to
> the running job, misunderstanding happens. As for our job, the
> misunderstanding results in forever stuck.
> The more details are as follows. The following figure shows how executor is
> allocated when the job starts to run. Now suppose only one executor is
> needed. In the figure, step 1,2,3 show how the executor is allocated. After
> the executor is allocated, it needs to register with the driver (step 4) and
> the driver responses to it (step 5). After the 5 steps, the executor can be
> used to run tasks.
> !How new executor is registered.png!
> In YarnAllocator, when step 3 is finished, it will increase the the variable
> (numExecutorsRunning) as show in the following code.
> {code:java}
> def updateInternalState(): Unit = synchronized {
> numExecutorsRunning += 1
> executorIdToContainer(executorId) = container
> containerIdToExecutorId(container.getId) = executorId
> val containerSet =
> allocatedHostToContainersMap.getOrElseUpdate(executorHostname,
> new HashSet[ContainerId])
> containerSet += containerId
> allocatedContainerToHostMap.put(containerId, executorHostname)
> }
> if (numExecutorsRunning < targetNumExecutors) {
> if (launchContainers) {
> launcherPool.execute(new Runnable {
> override def run(): Unit = {
> try {
> new ExecutorRunnable(
> Some(container),
> conf,
> sparkConf,
> driverUrl,
> executorId,
> executorHostname,
> executorMemory,
> executorCores,
> appAttemptId.getApplicationId.toString,
> securityMgr,
> localResources
> ).run()
> updateInternalState()
> } catch {
> case NonFatal(e) =>
> logError(s"Failed to launch executor $executorId on
> container $containerId", e)
> // Assigned container should be released immediately to
> avoid unnecessary resource
> // occupation.
> amClient.releaseAssignedContainer(containerId)
> }
> }
> })
> } else {
> // For test only
> updateInternalState()
> }
> } else {
> logInfo(("Skip launching executorRunnable as runnning Excecutors
> count: %d " +
> "reached target Executors count: %d.").format(numExecutorsRunning,
> targetNumExecutors))
> }
> {code}
> Imagine the step 4 is failed due to some reason (for example network
> fluctuation).
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
