Spark heuristic modification (#407)

* Revert "Dr. Elephant Tez Support working patch (#313)"

This reverts commit a0470a3.

* Rerevert "Dr. Elephant Tez Support working patch (#313)" including attribution.

This reverts commit e3fd598.

Co-authored-by: Abhishek Das <[email protected]>

* Auto tuning: Support for parameter set multi-try (#386)

* Changes in some of the Spark Heuristics

* Adding test for changes executor gc heuristic and unified memory heuristic

* Update ExecutorGcHeuristic.scala

* Update UnifiedMemoryHeuristic.scala

* Changed some hard coded values to variables

* Due to strict inequality changing the other thereshold levels for executor and driver

Author: ShubhamGupta29

app/com/linkedin/drelephant/ElephantContext.java

@@ -110,7 +110,6 @@ private void loadConfiguration() {
     loadFetchers();
     loadHeuristics();
     loadJobTypes();
-
     loadGeneralConf();
     loadAutoTuningConf();
 

app/com/linkedin/drelephant/spark/fetchers/SparkRestClient.scala

@@ -101,6 +101,7 @@ class SparkRestClient(sparkConf: SparkConf) {
         Await.result(futureJobDatas, DEFAULT_TIMEOUT),
         Await.result(futureStageDatas, DEFAULT_TIMEOUT),
         Await.result(futureExecutorSummaries, Duration(5, SECONDS)),
+        Seq.empty,
         Await.result(futureLogData, Duration(5, SECONDS))
       )
 

app/com/linkedin/drelephant/spark/heuristics/ConfigurationHeuristic.scala

@@ -118,7 +118,7 @@ class ConfigurationHeuristic(private val heuristicConfigurationData: HeuristicCo
       result.addResultDetail("Executor Overhead Memory", "Please do not specify excessive amount of overhead memory for Executors. Change it in the field " + SPARK_YARN_EXECUTOR_MEMORY_OVERHEAD)
     }
     if(evaluator.severityExecutorCores != Severity.NONE) {
-      result.addResultDetail("Executor cores", "The number of executor cores should be <=" + evaluator.DEFAULT_SPARK_CORES_THRESHOLDS.low + ". Please change it in the field " + SPARK_EXECUTOR_CORES_KEY)
+      result.addResultDetail("Executor cores", "The number of executor cores should be <" + evaluator.DEFAULT_SPARK_CORES_THRESHOLDS.low + ". Please change it in the field " + SPARK_EXECUTOR_CORES_KEY)
     }
     if(evaluator.severityExecutorMemory != Severity.NONE) {
       result.addResultDetail("Executor memory", "Please do not specify excessive amount of executor memory. Change it in the field " + SPARK_EXECUTOR_MEMORY_KEY)
@@ -204,7 +204,7 @@ object ConfigurationHeuristic {
       SeverityThresholds(low = MemoryFormatUtils.stringToBytes("10G"), MemoryFormatUtils.stringToBytes("15G"),
         severe = MemoryFormatUtils.stringToBytes("20G"), critical = MemoryFormatUtils.stringToBytes("25G"), ascending = true)
     val DEFAULT_SPARK_CORES_THRESHOLDS =
-      SeverityThresholds(low = 4, moderate = 6, severe = 8, critical = 10, ascending = true)
+      SeverityThresholds(low = 5, moderate = 7, severe = 9, critical = 11, ascending = true)
 
     val severityExecutorMemory = DEFAULT_SPARK_MEMORY_THRESHOLDS.severityOf(executorMemoryBytes.getOrElse(0).asInstanceOf[Number].longValue)
     val severityExecutorCores = DEFAULT_SPARK_CORES_THRESHOLDS.severityOf(executorCores.getOrElse(0).asInstanceOf[Number].intValue)

app/com/linkedin/drelephant/spark/heuristics/DriverHeuristic.scala

@@ -61,9 +61,10 @@ class DriverHeuristic(private val heuristicConfigurationData: HeuristicConfigura
         SPARK_DRIVER_MEMORY_KEY,
         formatProperty(evaluator.driverMemoryBytes.map(MemoryFormatUtils.bytesToString))
       ),
-      new HeuristicResultDetails(
-        "Ratio of time spent in GC to total time", evaluator.ratio.toString
-      ),
+      //Removing driver GC heuristics for now
+//      new HeuristicResultDetails(
+//        "Ratio of time spent in GC to total time", evaluator.ratio.toString
+//      ),
       new HeuristicResultDetails(
         SPARK_DRIVER_CORES_KEY,
         formatProperty(evaluator.driverCores.map(_.toString))
@@ -76,6 +77,7 @@ class DriverHeuristic(private val heuristicConfigurationData: HeuristicConfigura
     )
     if(evaluator.severityJvmUsedMemory != Severity.NONE) {
       resultDetails = resultDetails :+ new HeuristicResultDetails("Driver Peak JVM used Memory", "The allocated memory for the driver (in " + SPARK_DRIVER_MEMORY_KEY + ") is much more than the peak JVM used memory by the driver.")
+      resultDetails = resultDetails :+ new HeuristicResultDetails("Suggested spark.driver.memory", MemoryFormatUtils.roundOffMemoryStringToNextInteger(MemoryFormatUtils.bytesToString(((1 + BUFFER_FRACTION)  * (evaluator.maxDriverPeakJvmUsedMemory + reservedMemory)).toLong)))
     }
     if (evaluator.severityGc != Severity.NONE) {
       resultDetails = resultDetails :+ new HeuristicResultDetails("Gc ratio high", "The driver is spending too much time on GC. We recommend increasing the driver memory.")
@@ -113,6 +115,7 @@ object DriverHeuristic {
   val EXECUTION_MEMORY = "executionMemory"
   val STORAGE_MEMORY = "storageMemory"
   val JVM_USED_MEMORY = "jvmUsedMemory"
+  val BUFFER_FRACTION = 0.2
 
   // 300 * FileUtils.ONE_MB (300 * 1024 * 1024)
   val reservedMemory : Long = 314572800
@@ -172,10 +175,10 @@ object DriverHeuristic {
 
     //The following thresholds are for checking if the memory and cores values (driver) are above normal. These thresholds are experimental, and may change in the future.
     val DEFAULT_SPARK_MEMORY_THRESHOLDS =
-      SeverityThresholds(low = MemoryFormatUtils.stringToBytes("10G"), MemoryFormatUtils.stringToBytes("15G"),
+      SeverityThresholds(low = MemoryFormatUtils.stringToBytes("10G"), moderate = MemoryFormatUtils.stringToBytes("15G"),
         severe = MemoryFormatUtils.stringToBytes("20G"), critical = MemoryFormatUtils.stringToBytes("25G"), ascending = true)
     val DEFAULT_SPARK_CORES_THRESHOLDS =
-      SeverityThresholds(low = 4, moderate = 6, severe = 8, critical = 10, ascending = true)
+      SeverityThresholds(low = 5, moderate = 7, severe = 9, critical = 11, ascending = true)
 
     val severityDriverMemory = DEFAULT_SPARK_MEMORY_THRESHOLDS.severityOf(driverMemoryBytes.getOrElse(0).asInstanceOf[Number].longValue)
     val severityDriverCores = DEFAULT_SPARK_CORES_THRESHOLDS.severityOf(driverCores.getOrElse(0).asInstanceOf[Number].intValue)

app/com/linkedin/drelephant/spark/heuristics/ExecutorGcHeuristic.scala

@@ -59,7 +59,7 @@ class ExecutorGcHeuristic(private val heuristicConfigurationData: HeuristicConfi
     }
     //severityTimeD corresponds to the descending severity calculation
     if (evaluator.severityTimeD.getValue > Severity.LOW.getValue) {
-      resultDetails = resultDetails :+ new HeuristicResultDetails("Gc ratio low", "The job is spending too less time in GC. Please check if you have asked for more executor memory than required.")
+      resultDetails = resultDetails :+ new HeuristicResultDetails("Gc ratio low", "The job is spending too little time in GC. Please check if you have asked for more executor memory than required.")
     }
 
     val result = new HeuristicResult(
@@ -76,6 +76,7 @@ class ExecutorGcHeuristic(private val heuristicConfigurationData: HeuristicConfi
 object ExecutorGcHeuristic {
   val SPARK_EXECUTOR_MEMORY = "spark.executor.memory"
   val SPARK_EXECUTOR_CORES = "spark.executor.cores"
+  val EXECUTOR_RUNTIME_THRESHOLD_IN_MINUTES = 5
 
   /** The ascending severity thresholds for the ratio of JVM GC Time and executor Run Time (checking whether ratio is above normal)
     * These thresholds are experimental and are likely to change */
@@ -107,7 +108,12 @@ object ExecutorGcHeuristic {
 
     var ratio: Double = jvmTime.toDouble / executorRunTimeTotal.toDouble
 
-    lazy val severityTimeA: Severity = executorGcHeuristic.gcSeverityAThresholds.severityOf(ratio)
+    //If the total Executor Runtime is less then 5 minutes then we won't consider for the severity due to GC
+    lazy val severityTimeA: Severity = if (executorRunTimeTotal >= (EXECUTOR_RUNTIME_THRESHOLD_IN_MINUTES * Statistics.MINUTE_IN_MS))
+        executorGcHeuristic.gcSeverityAThresholds.severityOf(ratio)
+    else
+        Severity.NONE
+
     lazy val severityTimeD: Severity = executorGcHeuristic.gcSeverityDThresholds.severityOf(ratio)
 
     /**
@@ -145,4 +151,3 @@ object ExecutorGcHeuristic {
     }
   }
 }
-

app/com/linkedin/drelephant/spark/heuristics/UnifiedMemoryHeuristic.scala

@@ -72,6 +72,8 @@ object UnifiedMemoryHeuristic {
   val PEAK_UNIFIED_MEMORY_THRESHOLD_KEY = "peak_unified_memory_threshold"
   val SPARK_EXECUTOR_MEMORY_THRESHOLD_KEY = "spark_executor_memory_threshold"
   val DEFAULT_SPARK_EXECUTOR_MEMORY_THRESHOLD = "2G"
+  val UNIFIED_MEMORY_ALLOCATED_THRESHOLD = "256M"
+  val SPARK_MEMORY_FRACTION_THRESHOLD : Double = 0.05
 
   class Evaluator(unifiedMemoryHeuristic: UnifiedMemoryHeuristic, data: SparkApplicationData) {
     lazy val appConfigurationProperties: Map[String, String] =
@@ -116,10 +118,15 @@ object UnifiedMemoryHeuristic {
       }
     }.max
 
-    lazy val severity: Severity = if (sparkExecutorMemory <= MemoryFormatUtils.stringToBytes(unifiedMemoryHeuristic.sparkExecutorMemoryThreshold)) {
-      Severity.NONE
+    //If sparkMemoryFraction or total Unified Memory allocated is less than their respective thresholds then won't consider for severity
+     lazy val severity: Severity = if (sparkMemoryFraction > SPARK_MEMORY_FRACTION_THRESHOLD && maxMemory > MemoryFormatUtils.stringToBytes(UNIFIED_MEMORY_ALLOCATED_THRESHOLD)) {
+      if (sparkExecutorMemory <= MemoryFormatUtils.stringToBytes(unifiedMemoryHeuristic.sparkExecutorMemoryThreshold)) {
+        Severity.NONE
+      } else {
+        PEAK_UNIFIED_MEMORY_THRESHOLDS.severityOf(maxUnifiedMemory)
+      }
     } else {
-      PEAK_UNIFIED_MEMORY_THRESHOLDS.severityOf(maxUnifiedMemory)
+      Severity.NONE
     }
   }
 }

test/com/linkedin/drelephant/spark/heuristics/ConfigurationHeuristicTest.scala

@@ -143,7 +143,7 @@ class ConfigurationHeuristicTest extends FunSpec with Matchers {
       it("returns executor cores") {
         val details = heuristicResultDetails.get(10)
         details.getName should include("Executor cores")
-        details.getValue should be("The number of executor cores should be <=4. Please change it in the field spark.executor.cores")
+        details.getValue should be("The number of executor cores should be <5. Please change it in the field spark.executor.cores")
       }
     }
 

test/com/linkedin/drelephant/spark/heuristics/ExecutorGcHeuristicTest.scala

@@ -67,18 +67,57 @@ class ExecutorGcHeuristicTest extends FunSpec with Matchers {
       )
     )
 
+    val executorSummaries2 = Seq(
+      newFakeExecutorSummary(
+        id = "1",
+        totalGCTime = 12000,
+        totalDuration = Duration("4min").toMillis
+      ),
+      newFakeExecutorSummary(
+        id = "2",
+        totalGCTime = 13000,
+        totalDuration = Duration("1min").toMillis
+      )
+    )
+
+    val executorSummaries3 = Seq(
+      newFakeExecutorSummary(
+        id = "1",
+        totalGCTime = 9000,
+        totalDuration = Duration("2min").toMillis
+      )
+    )
+
     describe(".apply") {
       val data = newFakeSparkApplicationData(executorSummaries)
       val data1 = newFakeSparkApplicationData(executorSummaries1)
+      val data2 = newFakeSparkApplicationData(executorSummaries2)
+      val data3 = newFakeSparkApplicationData(executorSummaries3)
       val heuristicResult = executorGcHeuristic.apply(data)
       val heuristicResult1 = executorGcHeuristic.apply(data1)
+      val heuristicResult2 = executorGcHeuristic.apply(data2)
+      val heuristicResult3 = executorGcHeuristic.apply(data3)
       val heuristicResultDetails = heuristicResult.getHeuristicResultDetails
       val heuristicResultDetails1 = heuristicResult1.getHeuristicResultDetails
+      val heuristicResultDetails2 = heuristicResult2.getHeuristicResultDetails
+      val heuristicResultDetails3 = heuristicResult3.getHeuristicResultDetails
 
       it("returns the severity") {
         heuristicResult.getSeverity should be(Severity.CRITICAL)
       }
 
+      it("return the low severity") {
+        heuristicResult2.getSeverity should be(Severity.LOW)
+      }
+
+      it("return NONE severity for runtime less than 5 min") {
+        heuristicResult2.getSeverity should be(Severity.LOW)
+      }
+
+      it("return none severity") {
+        heuristicResult3.getSeverity should be(Severity.NONE)
+      }
+
       it("returns the JVM GC time to Executor Run time duration") {
         val details = heuristicResultDetails.get(0)
         details.getName should include("GC time to Executor Run time ratio")

test/com/linkedin/drelephant/spark/heuristics/UnifiedMemoryHeuristicTest.scala

@@ -21,6 +21,7 @@ class UnifiedMemoryHeuristicTest extends FunSpec with Matchers {
   val unifiedMemoryHeuristic = new UnifiedMemoryHeuristic(heuristicConfigurationData)
   val appConfigurationProperties = Map("spark.executor.memory"->"3147483647")
   val appConfigurationProperties1 = Map("spark.executor.memory"->"214567874847")
+  val appConfigurationProperties2 = Map("spark.executor.memory"->"214567874847", "spark.memory.fraction"->"0.06")
 
   val executorData = Seq(
     newDummyExecutorData("1", 999999999, Map("executionMemory" -> 300000, "storageMemory" -> 94567)),
@@ -41,13 +42,27 @@ class UnifiedMemoryHeuristicTest extends FunSpec with Matchers {
     newDummyExecutorData("2", 999999999, Map("executionMemory" -> 999999990, "storageMemory" -> 9))
   )
 
+  val executorData3 = Seq(
+    newDummyExecutorData("1", 400000, Map("executionMemory" -> 300000, "storageMemory" -> 94567)),
+    newDummyExecutorData("2", 500000, Map("executionMemory" -> 5000, "storageMemory" -> 9))
+  )
+
+  val executorData4 = Seq(
+    newDummyExecutorData("1", 268435460L, Map("executionMemory" -> 300000, "storageMemory" -> 94567)),
+    newDummyExecutorData("2", 268435460L, Map("executionMemory" -> 900000, "storageMemory" -> 500000))
+  )
+
   describe(".apply") {
     val data = newFakeSparkApplicationData(appConfigurationProperties, executorData)
     val data1 = newFakeSparkApplicationData(appConfigurationProperties1, executorData1)
     val data2 = newFakeSparkApplicationData(appConfigurationProperties1, executorData2)
+    val data3 = newFakeSparkApplicationData(appConfigurationProperties1, executorData3)
+    val data4 = newFakeSparkApplicationData(appConfigurationProperties2, executorData4)
     val heuristicResult = unifiedMemoryHeuristic.apply(data)
     val heuristicResult1 = unifiedMemoryHeuristic.apply(data1)
     val heuristicResult2 = unifiedMemoryHeuristic.apply(data2)
+    val heuristicResult3 = unifiedMemoryHeuristic.apply(data3)
+    val heuristicResult4 = unifiedMemoryHeuristic.apply(data4)
     val evaluator = new Evaluator(unifiedMemoryHeuristic, data1)
 
     it("has severity") {
@@ -85,6 +100,14 @@ class UnifiedMemoryHeuristicTest extends FunSpec with Matchers {
     it("has no severity when max and allocated memory are the same") {
       heuristicResult2.getSeverity should be(Severity.NONE)
     }
+
+    it("has no severity when maxMemory is less than 256Mb") {
+      heuristicResult3.getSeverity should be(Severity.NONE)
+    }
+
+    it("has critical severity when maxMemory is greater than 256Mb and spark memory fraction is greater than 0.05") {
+      heuristicResult4.getSeverity should be(Severity.CRITICAL)
+    }
   }
 }