Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/14079#discussion_r71411885
--- Diff:
core/src/main/scala/org/apache/spark/scheduler/BlacklistTracker.scala ---
@@ -0,0 +1,221 @@
+/*
+ * 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 java.util.concurrent.atomic.AtomicReference
+
+import scala.collection.mutable.{HashMap, HashSet}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config
+import org.apache.spark.util.Clock
+import org.apache.spark.util.SystemClock
+import org.apache.spark.util.Utils
+
+/**
+ * BlacklistTracker is designed to track problematic executors and nodes.
It supports blacklisting
+ * specific (executor, task) pairs within a stage, blacklisting entire
executors and nodes for a
+ * stage, and blacklisting executors and nodes across an entire
application (with a periodic
+ * expiry).
+ *
+ * The tracker needs to deal with a variety of workloads, eg.: bad user
code, which may lead to many
+ * task failures, but that should not count against individual executors;
many small stages, which
+ * may prevent a bad executor for having many failures within one stage,
but still many failures
+ * over the entire application; "flaky" executors, that don't fail every
task, but are still
+ * faulty; etc.
+ *
+ * THREADING: As with most helpers of TaskSchedulerImpl, this is not
thread-safe. Though it is
+ * called by multiple threads, callers must already have a lock on the
TaskSchedulerImpl. The
+ * one exception is [[nodeBlacklist()]], which can be called without
holding a lock.
+ */
+private[scheduler] class BlacklistTracker (
+ conf: SparkConf,
+ clock: Clock = new SystemClock()) extends Logging {
+
+ private val MAX_FAILURES_PER_EXEC =
conf.get(config.MAX_FAILURES_PER_EXEC)
+ private val MAX_FAILED_EXEC_PER_NODE =
conf.get(config.MAX_FAILED_EXEC_PER_NODE)
+ val EXECUTOR_RECOVERY_MILLIS =
BlacklistTracker.getBlacklistExpiryTime(conf)
+
+ // a count of failed tasks for each executor. Only counts failures
after tasksets complete
+ // successfully
+ private val executorIdToFailureCount: HashMap[String, Int] = HashMap()
+ private val executorIdToBlacklistExpiryTime: HashMap[String, Long] = new
HashMap()
+ private val nodeIdToBlacklistExpiryTime: HashMap[String, Long] = new
HashMap()
+ private val _nodeBlacklist: AtomicReference[Set[String]] = new
AtomicReference(Set())
+ private var nextExpiryTime: Long = Long.MaxValue
+
+ def start(): Unit = {}
+
+ def stop(): Unit = {}
+
+ def expireExecutorsInBlacklist(): Unit = {
+ val now = clock.getTimeMillis()
+ // quickly check if we've got anything to expire from blacklist -- if
not, avoid doing any work
+ if (now > nextExpiryTime) {
+ val execsToClear = executorIdToBlacklistExpiryTime.filter(_._2 <
now).keys
+ if (execsToClear.nonEmpty) {
+ logInfo(s"Removing executors $execsToClear from blacklist during
periodic recovery")
+ execsToClear.foreach { exec =>
executorIdToBlacklistExpiryTime.remove(exec) }
+ }
+ if (executorIdToBlacklistExpiryTime.nonEmpty) {
+ nextExpiryTime = executorIdToBlacklistExpiryTime.map{_._2}.min
+ } else {
+ nextExpiryTime = Long.MaxValue
+ }
+ val nodesToClear = nodeIdToBlacklistExpiryTime.filter(_._2 <
now).keys
+ if (nodesToClear.nonEmpty) {
+ logInfo(s"Removing nodes $nodesToClear from blacklist during
periodic recovery")
+ nodesToClear.foreach { node =>
nodeIdToBlacklistExpiryTime.remove(node) }
+ // make a copy of the blacklisted nodes so nodeBlacklist() is
threadsafe
+ _nodeBlacklist.set(nodeIdToBlacklistExpiryTime.keySet.toSet)
+ }
+ }
+ }
+
+ def taskSetSucceeded(
+ failuresByExec: HashMap[String, FailureStatus],
+ scheduler: TaskSchedulerImpl): Unit = {
+ // if any tasks failed, we count them towards the overall failure
count for the executor at
+ // this point.
+ failuresByExec.foreach { case (exec, newFailures) =>
+ val prevFailures = executorIdToFailureCount.getOrElse(exec, 0)
+ val newTotal = prevFailures + newFailures.totalFailures
+
+ if (newTotal >= MAX_FAILURES_PER_EXEC) {
+ logInfo(s"Blacklisting executor $exec because it has $newTotal" +
+ s" task failures in successful task sets")
+ val now = clock.getTimeMillis()
+ val expiryTime = now + EXECUTOR_RECOVERY_MILLIS
+ executorIdToBlacklistExpiryTime.put(exec, expiryTime)
+ executorIdToFailureCount.remove(exec)
+ if (expiryTime < nextExpiryTime) {
+ nextExpiryTime = expiryTime
+ }
+
+ val node = scheduler.getHostForExecutor(exec)
+ val execs =
scheduler.getExecutorsAliveOnHost(node).getOrElse(Set())
--- End diff --
oh good point. this design was somewhat intentional -- to deal w/
intermittent flakiness, you wouldn't blacklist a node if eg. you had one
executor get blacklisted on it 3 times, but with long periods in between. I
see though that this is probably undesirable with DA.
Its actually a really good point, that with DA node blacklisting is
important even with just one executor per node.
We could either:
1) keep it as it is now, so each new executor that DA adds has to get
re-blacklisted.
2) always blacklist the node after N blacklisted executors, regardless of
the time between
3) add a timeout for how much time can go between blacklisted executors for
it to count. Then we've got to either add another configuration or re-use one
of the existing ones.
I don't like the added configuration burden of 3 ... I'm leaning towards 2.
Though maybe I'm forced to use 3 just because otherwise we track failed
executors for all time, which I guess would be a memory leak, though hopefully
an extremely slow one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
