-
Notifications
You must be signed in to change notification settings - Fork 28k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[SPARK-17675][CORE] Expand Blacklist for TaskSets
## What changes were proposed in this pull request? This is a step along the way to SPARK-8425. To enable incremental review, the first step proposed here is to expand the blacklisting within tasksets. In particular, this will enable blacklisting for * (task, executor) pairs (this already exists via an undocumented config) * (task, node) * (taskset, executor) * (taskset, node) Adding (task, node) is critical to making spark fault-tolerant of one-bad disk in a cluster, without requiring careful tuning of "spark.task.maxFailures". The other additions are also important to avoid many misleading task failures and long scheduling delays when there is one bad node on a large cluster. Note that some of the code changes here aren't really required for just this -- they put pieces in place for SPARK-8425 even though they are not used yet (eg. the `BlacklistTracker` helper is a little out of place, `TaskSetBlacklist` holds onto a little more info than it needs to for just this change, and `ExecutorFailuresInTaskSet` is more complex than it needs to be). ## How was this patch tested? Added unit tests, run tests via jenkins. Author: Imran Rashid <irashid@cloudera.com> Author: mwws <wei.mao@intel.com> Closes #15249 from squito/taskset_blacklist_only.
- Loading branch information
Showing
17 changed files
with
964 additions
and
198 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
114 changes: 114 additions & 0 deletions
114
core/src/main/scala/org/apache/spark/scheduler/BlacklistTracker.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.scheduler | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.internal.config | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| private[scheduler] object BlacklistTracker extends Logging { | ||
|
|
||
| private val DEFAULT_TIMEOUT = "1h" | ||
|
|
||
| /** | ||
| * Returns true if the blacklist is enabled, based on checking the configuration in the following | ||
| * order: | ||
| * 1. Is it specifically enabled or disabled? | ||
| * 2. Is it enabled via the legacy timeout conf? | ||
| * 3. Default is off | ||
| */ | ||
| def isBlacklistEnabled(conf: SparkConf): Boolean = { | ||
| conf.get(config.BLACKLIST_ENABLED) match { | ||
| case Some(enabled) => | ||
| enabled | ||
| case None => | ||
| // if they've got a non-zero setting for the legacy conf, always enable the blacklist, | ||
| // otherwise, use the default. | ||
| val legacyKey = config.BLACKLIST_LEGACY_TIMEOUT_CONF.key | ||
| conf.get(config.BLACKLIST_LEGACY_TIMEOUT_CONF).exists { legacyTimeout => | ||
| if (legacyTimeout == 0) { | ||
| logWarning(s"Turning off blacklisting due to legacy configuration: $legacyKey == 0") | ||
| false | ||
| } else { | ||
| logWarning(s"Turning on blacklisting due to legacy configuration: $legacyKey > 0") | ||
| true | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def getBlacklistTimeout(conf: SparkConf): Long = { | ||
| conf.get(config.BLACKLIST_TIMEOUT_CONF).getOrElse { | ||
| conf.get(config.BLACKLIST_LEGACY_TIMEOUT_CONF).getOrElse { | ||
| Utils.timeStringAsMs(DEFAULT_TIMEOUT) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Verify that blacklist configurations are consistent; if not, throw an exception. Should only | ||
| * be called if blacklisting is enabled. | ||
| * | ||
| * The configuration for the blacklist is expected to adhere to a few invariants. Default | ||
| * values follow these rules of course, but users may unwittingly change one configuration | ||
| * without making the corresponding adjustment elsewhere. This ensures we fail-fast when | ||
| * there are such misconfigurations. | ||
| */ | ||
| def validateBlacklistConfs(conf: SparkConf): Unit = { | ||
|
|
||
| def mustBePos(k: String, v: String): Unit = { | ||
| throw new IllegalArgumentException(s"$k was $v, but must be > 0.") | ||
| } | ||
|
|
||
| Seq( | ||
| config.MAX_TASK_ATTEMPTS_PER_EXECUTOR, | ||
| config.MAX_TASK_ATTEMPTS_PER_NODE, | ||
| config.MAX_FAILURES_PER_EXEC_STAGE, | ||
| config.MAX_FAILED_EXEC_PER_NODE_STAGE | ||
| ).foreach { config => | ||
| val v = conf.get(config) | ||
| if (v <= 0) { | ||
| mustBePos(config.key, v.toString) | ||
| } | ||
| } | ||
|
|
||
| val timeout = getBlacklistTimeout(conf) | ||
| if (timeout <= 0) { | ||
| // first, figure out where the timeout came from, to include the right conf in the message. | ||
| conf.get(config.BLACKLIST_TIMEOUT_CONF) match { | ||
| case Some(t) => | ||
| mustBePos(config.BLACKLIST_TIMEOUT_CONF.key, timeout.toString) | ||
| case None => | ||
| mustBePos(config.BLACKLIST_LEGACY_TIMEOUT_CONF.key, timeout.toString) | ||
| } | ||
| } | ||
|
|
||
| val maxTaskFailures = conf.get(config.MAX_TASK_FAILURES) | ||
| val maxNodeAttempts = conf.get(config.MAX_TASK_ATTEMPTS_PER_NODE) | ||
|
|
||
| if (maxNodeAttempts >= maxTaskFailures) { | ||
| throw new IllegalArgumentException(s"${config.MAX_TASK_ATTEMPTS_PER_NODE.key} " + | ||
| s"( = ${maxNodeAttempts}) was >= ${config.MAX_TASK_FAILURES.key} " + | ||
| s"( = ${maxTaskFailures} ). Though blacklisting is enabled, with this configuration, " + | ||
| s"Spark will not be robust to one bad node. Decrease " + | ||
| s"${config.MAX_TASK_ATTEMPTS_PER_NODE.key}, increase ${config.MAX_TASK_FAILURES.key}, " + | ||
| s"or disable blacklisting with ${config.BLACKLIST_ENABLED.key}") | ||
| } | ||
| } | ||
| } |
50 changes: 50 additions & 0 deletions
50
core/src/main/scala/org/apache/spark/scheduler/ExecutorFailuresInTaskSet.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.spark.scheduler | ||
|
|
||
| import scala.collection.mutable.HashMap | ||
|
|
||
| /** | ||
| * Small helper for tracking failed tasks for blacklisting purposes. Info on all failures on one | ||
| * executor, within one task set. | ||
| */ | ||
| private[scheduler] class ExecutorFailuresInTaskSet(val node: String) { | ||
| /** | ||
| * Mapping from index of the tasks in the taskset, to the number of times it has failed on this | ||
| * executor. | ||
| */ | ||
| val taskToFailureCount = HashMap[Int, Int]() | ||
|
|
||
| def updateWithFailure(taskIndex: Int): Unit = { | ||
| val prevFailureCount = taskToFailureCount.getOrElse(taskIndex, 0) | ||
| taskToFailureCount(taskIndex) = prevFailureCount + 1 | ||
| } | ||
|
|
||
| def numUniqueTasksWithFailures: Int = taskToFailureCount.size | ||
|
|
||
| /** | ||
| * Return the number of times this executor has failed on the given task index. | ||
| */ | ||
| def getNumTaskFailures(index: Int): Int = { | ||
| taskToFailureCount.getOrElse(index, 0) | ||
| } | ||
|
|
||
| override def toString(): String = { | ||
| s"numUniqueTasksWithFailures = $numUniqueTasksWithFailures; " + | ||
| s"tasksToFailureCount = $taskToFailureCount" | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.