LIHADOOP-39635: Add new configuration parameters heuristic #463

edwinalu · 2018-10-30T13:20:11Z

Add new configuration parameters heuristic, which will list the current values for configuration parameters, and also recommended new values. To determine new values, it will check for:

execution memory spill: this will slow down the application, so try to prevent this by increasing partitions, increasing executor memory, or decreasing cores.
long tasks: this will slow down the application, so try to prevent this by increasing the number of partitions.
task skew: this will slow down the application, so add recommendations for making partitions more even.
OOM or GC: increase memory, increase partitions, or decrease cores, to try to avoid the error.
container killed by YARN errors: increase overhead memory.
unused executor memory, if this is much higher than max JVM used memory; either increase cores or decrease memory.
driver configuration parameters (memory and cores).

pralabhkumar · 2018-12-04T09:45:08Z

#438 have my comments ,which are already resolved in this pull request

pralabhkumar · 2018-12-04T09:49:58Z

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

+       // check if executor memory can be lowered
+         adjustExecutorMemory()
+     }
+


Whats the rational behind calling first adjustParametersForLongTasks, adjustParametersForExecutionMemorySpill, adjustParametersForGCandOOM . Is there a priority defined to fix long running task and then memory spill

The ordering is somewhat arbitrary, but there is a better idea of how much to increase the number of partitions for long tasks, and a reasonable estimate for how much to adjust memory/cores/partitions for execution memory spill. GC and OOM are just a guess. It is going from more exact to less exact.

The adjustment(s) for each could affect the other conditions as well (if more partitions are specified due to long tasks, then this would also help with execution memory spill, and OOM/GC).

pralabhkumar · 2018-12-04T10:00:34Z

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

+       * If so, either increase cores to make better use of the memory, or decrease executor
+       * memory.
+       */
+     private def adjustExecutorMemory() = {


Decreasing executor memory may cause more spill , how are we handling the same scenario

If there's execution spill, then there shouldn't be too memory allocated. However, it makes sense to be cautious. I'll add a check for execution spill as well, and not adjust the memory in this case.

pralabhkumar · 2018-12-04T10:03:48Z

@edwinalu How are you planning to integrate with Unified Architecture / TuneIn

edwinalu · 2018-12-04T23:02:06Z

@pralabhkumar , yes, I am not sure what would be the best way to integrate with TuneIn. When we discussed a few weeks ago, the idea was to merge the heuristic first. Let's discuss at the meeting. It would be good to keep in sync.

mkumar1984 · 2018-12-06T15:52:52Z

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

+       while (iter.hasNext && continueFn(modified)) {
+         iter.next() match {
+           case adjustment: CoreDivisorAdjustment =>
+             if (adjustment.canAdjust(recommendedExecutorCores)) {


If code for these two cases are similar, then one of them can be removed. Same is true for other cases as well.

It is processing the adjustment differently for each adjustment type. Are you suggesting creating more case classes, CoreAdjustment, MemoryAdjustment, and PartitionAdjustment, and then subclassing the current case classes off those, to consolidate?

This is only related to merging two cases if code is same for these 2 cases.
For example
Case 1:
//Code
Case 2:
//Code

Case 1 | 2 :
//Code:

I've added traits for CoreAdjustment and MemoryAdjustment, and consolidated the case classes.

mkumar1984 · 2018-12-06T16:01:19Z

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

+
+     val currentParallelism = sparkExecutorInstances.map(_ * sparkExecutorCores)
+
+     val jvmUsedMemoryHeuristic =


In the long run, we need to think about caching these heuristics, so that these need not be computed multiple times. For now this is fine.

mkumar1984 · 2018-12-06T17:52:09Z

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

+       * @return the recommended value in bytes for executor memory overhead
+       */
+     private def calculateExecutorMemoryOverhead(): Option[Long] = {
+       val overheadMemoryIncrement = 1L * GB_TO_BYTES


You can use FileUtils.ONE_GB.

mkumar1984 · 2018-12-06T17:55:00Z

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

+       if (stageAnalysis.exists { stage =>
+         hasSignificantSeverity(stage.taskFailureResult.containerKilledSeverity)
+       }) {
+         val actualMemoryOverhead = sparkExecutorMemoryOverhead.getOrElse {


Shouldn't we consider user specified memory overhead as well ?

It is first trying to get the user specified memory overhead, and if this doesn't exist, calculating the default value.

Missed that.

mkumar1984 · 2018-12-06T18:02:04Z

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

+           num * MB_TO_BYTES
+         } else {
+           unit.charAt(0) match {
+             case 'T' => num * TB_TO_BYTES


You can use FileUtils.ONE_GB, ONE_MB etc.

mkumar1984 · 2018-12-06T18:07:32Z

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

+     * @param size The memory value in long bytes
+     * @return The formatted string, null if
+     */
+   private def bytesToString(size: Long): String = {


This method is available in org.apache.spark.network.util.JavaUtils.

I've replaced JstringToBytes with avaUtils.byteStringAsBytes. I wasn't able to find a function for the other direction.

mkumar1984 · 2018-12-06T18:08:31Z

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

+
+import com.linkedin.drelephant.analysis.SeverityThresholds
+
+object ConfigurationUtils {


Would be good to change the name to have ConfigurationHeuristicsConstants or something else?

pralabhkumar · 2018-12-10T12:42:41Z

LGTM , please test the flow on pokemon/EI cluster to make sure things are working fine e2e

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

…ves. Do not print stack trace for fetching failed tasks.

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

varunsaxena · 2018-12-17T11:01:22Z

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

+     * @param size The memory value in long bytes
+     * @return The formatted string, null if
+     */
+   private def bytesToString(size: Long): String = {


MemoryFormatUtils#bytesToString does something similar. It won't print bytes in MB if its < 2GB but otherwise its exactly same. Maybe we can modify MemoryFormatUtils#bytesToString if this condition is necessary or move this method to MemoryFormatUtils too (As this method is good candidate for utils)

Other differences are that MemoryFormatUtils#bytesToString also have a space between the value and unit, so "2 GB" instead of "2GB", and it only does GB and MB (no KB, or B) and rounds up. It would be possible to modify MemoryFormatUtils#bytesToString to have additional parameters for specify the threshold for moving to the next unit, which units to use, if it should round up, and if it should add a space or not.

varunsaxena · 2018-12-17T13:25:23Z

app/com/linkedin/drelephant/spark/legacydata/SparkStageData.java

+
+/**
+ * This class contains Spark stage information.
+ */public class SparkStageData {


What's this class meant for?

Sorry, I'm not sure how this got added -- it is not being used. It may have somehow been a merge conflict when copying over from another branch. I'll remove.

varunsaxena · 2018-12-18T18:27:15Z

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

+    val (numTasksWithContainerKilled, containerKilledSeverity) =
+      checkForSpecificTaskError(stageId, stageData, failedTasks,
+        StagesWithFailedTasksHeuristic.OVERHEAD_MEMORY_ERROR,
+        "the container was killed by YARN for exeeding memory limits.", details)


Is this message correct? For instance, exeeding should ideally be "exceeding". Is the message generated in YarnAllocator in Spark code? If yes, the message may well be incorrect.
Moreover, going ahead, probably Spark can do error categorization by itself and pass a well-defined enum instead of Dr.Elephant expecting a custom message generated in Spark code because such code can break. Error message can still be passed as usual and printed to give user detailed information.

Yes, this should be "exceeded". For searching the actual error message, it is using StagesWithFailedTasksHeuristic.OVERHEAD_MEMORY_ERROR.

Right now, Spark is returning the error message, which can be varied, if it is coming from the user application. There isn't a well-defined enum for types of errors.

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

varunsaxena · 2018-12-18T19:17:42Z

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

+       data.appConfigurationProperties
+
+     // current configuration parameters
+     lazy val sparkExecutorMemory = JavaUtils.byteStringAsBytes(


Can't we use MemoryFormatUtils#stringToBytes which is in Dr.Elephant codebase instead of using Spark utils. It will do exactly the same thing. Utils are typically meant to be used within a project even though they are public classes.

Yes, this does seem to be the same, and I will change.

varunsaxena · 2018-12-18T19:40:19Z

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

@@ -211,7 +218,7 @@ class SparkRestClient(sparkConf: SparkConf) {
  }

  private def getStageDatas(attemptTarget: WebTarget): Seq[StageDataImpl] = {
-    val target = attemptTarget.path("stages")
+    val target = attemptTarget.path("stages/withSummaries")


This is LinkedIn specific REST endpoint and wont work in open source till it's contributed back to Spark upstream. Probably going ahead we should refactor the code and have our own SparkRestClient implementation. The abstraction for us is primarily at the fetcher level. So probably have a linkedin specific spark fetcher implementation which extends SparkFetcher which currently exists, reuses the part where we are fetching event logs but has custom Spark rest client implementation.

Yes, this is LinkedIn specific, and separating out the code would make sense. Could the refactoring be done later?

* Add new configuration parameters heuristic * add configuration * check for execution memory spill before adjusting executor memory * code review comments * remove partitions * consolidate case classes * add license * add more licenses * remove stage level GC analysis/warnings, due to too many false positives. Do not print stack trace for fetching failed tasks. * code review comments (cherry picked from commit 07c2446)

edwinalu added 2 commits October 30, 2018 06:17

Add new configuration parameters heuristic

05cc90b

add configuration

35c7ab4

pralabhkumar reviewed Dec 4, 2018

View reviewed changes

check for execution memory spill before adjusting executor memory

f976b0e

mkumar1984 reviewed Dec 6, 2018

View reviewed changes

edwinalu added 2 commits December 6, 2018 13:49

code review comments

4abc701

remove partitions

5bc6d6b

consolidate case classes

aae9c16

varunsaxena requested changes Dec 11, 2018

View reviewed changes

app/com/linkedin/drelephant/spark/heuristics/ConfigurationParameterAdjustment.scala Show resolved Hide resolved

edwinalu added 3 commits December 11, 2018 07:49

add license

73a3606

add more licenses

5010e41

remove stage level GC analysis/warnings, due to too many false positi…

4ba707b

…ves. Do not print stack trace for fetching failed tasks.

varunsaxena requested changes Dec 13, 2018

View reviewed changes

app/com/linkedin/drelephant/spark/heuristics/ConfigurationParametersHeuristic.scala Show resolved Hide resolved

varunsaxena requested changes Dec 18, 2018

View reviewed changes

code review comments

ca9d1f4

mkumar1984 merged commit 07c2446 into linkedin:tuning Dec 21, 2018


		val currentParallelism = sparkExecutorInstances.map(_ * sparkExecutorCores)

		val jvmUsedMemoryHeuristic =


		import com.linkedin.drelephant.analysis.SeverityThresholds

		object ConfigurationUtils {

LIHADOOP-39635: Add new configuration parameters heuristic #463

LIHADOOP-39635: Add new configuration parameters heuristic #463

Conversation

edwinalu commented Oct 30, 2018

pralabhkumar commented Dec 4, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pralabhkumar Dec 4, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pralabhkumar commented Dec 4, 2018

edwinalu commented Dec 4, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pralabhkumar commented Dec 10, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

varunsaxena Dec 18, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

varunsaxena Dec 18, 2018 • edited Loading

Choose a reason for hiding this comment

edwinalu Dec 19, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edwinalu Dec 19, 2018 • edited Loading

Choose a reason for hiding this comment

pralabhkumar Dec 4, 2018 •

edited

Loading

pralabhkumar commented Dec 10, 2018 •

edited

Loading

varunsaxena Dec 18, 2018 •

edited

Loading

varunsaxena Dec 18, 2018 •

edited

Loading

edwinalu Dec 19, 2018 •

edited

Loading

edwinalu Dec 19, 2018 •

edited

Loading