[SPARK-37819][K8S] Add OUTLIER executor roll policy and use it by default (apache#1323)

### What changes were proposed in this pull request?

This PR aims to add a new executor roll policy, `OUTLIER`, which aims to detect various outliers first, and use it by default.
If there is no outlier, it will work like `TOTAL_DURATION` policy.

### Why are the changes needed?

The users can use `OUTLIER` policy to consider the outliers in terms of multiple dimensions.
In addition, this will be a better default policy.

### Does this PR introduce _any_ user-facing change?

No. This is a new feature in Apache Spark 3.3.

### How was this patch tested?

Pass the CIs with the newly added test cases.

Author: dongjoon-hyun

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -147,26 +147,30 @@ private[spark] object Config extends Logging {
       .createWithDefault(0)
 
   object ExecutorRollPolicy extends Enumeration {
-    val ID, ADD_TIME, TOTAL_GC_TIME, TOTAL_DURATION, AVERAGE_DURATION, FAILED_TASKS = Value
+    val ID, ADD_TIME, TOTAL_GC_TIME, TOTAL_DURATION, AVERAGE_DURATION, FAILED_TASKS, OUTLIER = Value
   }
 
   val EXECUTOR_ROLL_POLICY =
     ConfigBuilder("spark.kubernetes.executor.rollPolicy")
-      .doc("Executor roll policy: Valid values are ID, ADD_TIME, TOTAL_GC_TIME (default), " +
-        "TOTAL_DURATION, and FAILED_TASKS. " +
+      .doc("Executor roll policy: Valid values are ID, ADD_TIME, TOTAL_GC_TIME, " +
+        "TOTAL_DURATION, FAILED_TASKS, and OUTLIER (default). " +
         "When executor roll happens, Spark uses this policy to choose " +
         "an executor and decommission it. The built-in policies are based on executor summary." +
         "ID policy chooses an executor with the smallest executor ID. " +
         "ADD_TIME policy chooses an executor with the smallest add-time. " +
         "TOTAL_GC_TIME policy chooses an executor with the biggest total task GC time. " +
         "TOTAL_DURATION policy chooses an executor with the biggest total task time. " +
         "AVERAGE_DURATION policy chooses an executor with the biggest average task time. " +
-        "FAILED_TASKS policy chooses an executor with the most number of failed tasks.")
+        "FAILED_TASKS policy chooses an executor with the most number of failed tasks. " +
+        "OUTLIER policy chooses an executor with outstanding statistics which is bigger than" +
+        "at least two standard deviation from the mean in average task time, " +
+        "total task time, total task GC time, and the number of failed tasks if exists. " +
+        "If there is no outlier, it works like TOTAL_DURATION policy.")
       .version("3.3.0")
       .stringConf
       .transform(_.toUpperCase(Locale.ROOT))
       .checkValues(ExecutorRollPolicy.values.map(_.toString))
-      .createWithDefault(ExecutorRollPolicy.TOTAL_GC_TIME.toString)
+      .createWithDefault(ExecutorRollPolicy.OUTLIER.toString)
 
   val MINIMUM_TASKS_PER_EXECUTOR_BEFORE_ROLLING =
     ConfigBuilder("spark.kubernetes.executor.minTasksPerExecutorBeforeRolling")

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorRollPlugin.scala

@@ -16,6 +16,7 @@
  */
 package org.apache.spark.scheduler.cluster.k8s
 
+import java.lang.Math.sqrt
 import java.util.{Map => JMap}
 import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
 
@@ -116,7 +117,42 @@ class ExecutorRollDriverPlugin extends DriverPlugin with Logging {
         listWithoutDriver.sortBy(e => e.totalDuration.toFloat / Math.max(1, e.totalTasks)).reverse
       case ExecutorRollPolicy.FAILED_TASKS =>
         listWithoutDriver.sortBy(_.failedTasks).reverse
+      case ExecutorRollPolicy.OUTLIER =>
+        // We build multiple outlier lists and concat in the following importance order to find
+        // outliers in various perspective:
+        //   AVERAGE_DURATION > TOTAL_DURATION > TOTAL_GC_TIME > FAILED_TASKS
+        // Since we will choose only first item, the duplication is okay. If there is no outlier,
+        // We fallback to TOTAL_DURATION policy.
+        outliers(listWithoutDriver.filter(_.totalTasks > 0), e => e.totalDuration / e.totalTasks) ++
+          outliers(listWithoutDriver, e => e.totalDuration) ++
+          outliers(listWithoutDriver, e => e.totalGCTime) ++
+          outliers(listWithoutDriver, e => e.failedTasks) ++
+          listWithoutDriver.sortBy(_.totalDuration).reverse
     }
     sortedList.headOption.map(_.id)
   }
+
+  /**
+   * Return executors whose metrics is outstanding, '(value - mean) > 2-sigma'. This is
+   * a best-effort approach because the snapshot of ExecutorSummary is not a normal distribution.
+   * Outliers can be defined in several ways (https://en.wikipedia.org/wiki/Outlier).
+   * Here, we borrowed 2-sigma idea from https://en.wikipedia.org/wiki/68-95-99.7_rule.
+   * In case of normal distribution, this is known to be 2.5 percent roughly.
+   */
+  private def outliers(
+      list: Seq[v1.ExecutorSummary],
+      get: v1.ExecutorSummary => Float): Seq[v1.ExecutorSummary] = {
+    if (list.isEmpty) {
+      list
+    } else {
+      val size = list.size
+      val values = list.map(get)
+      val mean = values.sum / size
+      val sd = sqrt(values.map(v => (v - mean) * (v - mean)).sum / size)
+      list
+        .filter(e => (get(e) - mean) > 2 * sd)
+        .sortBy(e => get(e))
+        .reverse
+    }
+  }
 }

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorRollPluginSuite.scala

@@ -42,7 +42,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithSmallestID = new ExecutorSummary("1", "host:port", true, 1,
     10, 10, 1, 1, 1,
     0, 0, 1, 100,
-    1, 100, 100,
+    20, 100, 100,
     10, false, 20, new Date(1639300001000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -51,7 +51,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithSmallestAddTime = new ExecutorSummary("2", "host:port", true, 1,
     10, 10, 1, 1, 1,
     0, 0, 1, 100,
-    1, 100, 100,
+    20, 100, 100,
     10, false, 20, new Date(1639300000000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -60,7 +60,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithBiggestTotalGCTime = new ExecutorSummary("3", "host:port", true, 1,
     10, 10, 1, 1, 1,
     0, 0, 1, 100,
-    4, 100, 100,
+    40, 100, 100,
     10, false, 20, new Date(1639300002000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -69,7 +69,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithBiggestTotalDuration = new ExecutorSummary("4", "host:port", true, 1,
     10, 10, 1, 1, 1,
     0, 0, 4, 400,
-    1, 100, 100,
+    20, 100, 100,
     10, false, 20, new Date(1639300003000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -78,7 +78,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithBiggestFailedTasks = new ExecutorSummary("5", "host:port", true, 1,
     10, 10, 1, 1, 1,
     5, 0, 1, 100,
-    1, 100, 100,
+    20, 100, 100,
     10, false, 20, new Date(1639300003000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -87,7 +87,7 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
   val execWithBiggestAverageDuration = new ExecutorSummary("6", "host:port", true, 1,
     10, 10, 1, 1, 1,
     0, 0, 2, 300,
-    1, 100, 100,
+    20, 100, 100,
     10, false, 20, new Date(1639300003000L),
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
@@ -101,9 +101,18 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
     Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
     false, Set())
 
+  // This is used to stabilize 'mean' and 'sd' in OUTLIER test cases.
+  val execNormal = new ExecutorSummary("8", "host:port", true, 1,
+    10, 10, 1, 1, 1,
+    4, 0, 2, 280,
+    30, 100, 100,
+    10, false, 20, new Date(1639300001000L),
+    Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
+    false, Set())
+
   val list = Seq(driverSummary, execWithSmallestID, execWithSmallestAddTime,
     execWithBiggestTotalGCTime, execWithBiggestTotalDuration, execWithBiggestFailedTasks,
-    execWithBiggestAverageDuration, execWithoutTasks)
+    execWithBiggestAverageDuration, execWithoutTasks, execNormal)
 
   override def beforeEach(): Unit = {
     super.beforeEach()
@@ -162,4 +171,49 @@ class ExecutorRollPluginSuite extends SparkFunSuite with PrivateMethodTester {
       Some("6"),
       plugin.invokePrivate(_choose(list, ExecutorRollPolicy.AVERAGE_DURATION)))
   }
+
+  test("Policy: OUTLIER - Work like TOTAL_DURATION if there is no outlier") {
+    assertEquals(
+      plugin.invokePrivate(_choose(list, ExecutorRollPolicy.TOTAL_DURATION)),
+      plugin.invokePrivate(_choose(list, ExecutorRollPolicy.OUTLIER)))
+  }
+
+  test("Policy: OUTLIER - Detect an average task duration outlier") {
+    val outlier = new ExecutorSummary("9999", "host:port", true, 1,
+      0, 0, 1, 0, 0,
+      3, 0, 1, 300,
+      20, 0, 0,
+      0, false, 0, new Date(1639300001000L),
+      Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
+      false, Set())
+    assertEquals(
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.AVERAGE_DURATION)),
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.OUTLIER)))
+  }
+
+  test("Policy: OUTLIER - Detect a total task duration outlier") {
+    val outlier = new ExecutorSummary("9999", "host:port", true, 1,
+      0, 0, 1, 0, 0,
+      3, 0, 1000, 1000,
+      0, 0, 0,
+      0, false, 0, new Date(1639300001000L),
+      Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
+      false, Set())
+    assertEquals(
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.TOTAL_DURATION)),
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.OUTLIER)))
+  }
+
+  test("Policy: OUTLIER - Detect a total GC time outlier") {
+    val outlier = new ExecutorSummary("9999", "host:port", true, 1,
+      0, 0, 1, 0, 0,
+      3, 0, 1, 100,
+      1000, 0, 0,
+      0, false, 0, new Date(1639300001000L),
+      Option.empty, Option.empty, Map(), Option.empty, Set(), Option.empty, Map(), Map(), 1,
+      false, Set())
+    assertEquals(
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.TOTAL_GC_TIME)),
+      plugin.invokePrivate(_choose(list :+ outlier, ExecutorRollPolicy.OUTLIER)))
+  }
 }