diff --git a/app-conf/HeuristicConf.xml b/app-conf/HeuristicConf.xml
index 1f75292c7..2c69e719a 100644
--- a/app-conf/HeuristicConf.xml
+++ b/app-conf/HeuristicConf.xml
@@ -422,4 +422,26 @@
-->
+
This is a heuristic for Configuration Parameters.
+This shows the values for the listed configuration parameters for this execution of the Spark application.
+This shows the recommended values for the listed configuration parameters, based on Dr. Elephant's analysis + of the metrics from this run.
diff --git a/test/com/linkedin/drelephant/spark/fetchers/SparkRestClientTest.scala b/test/com/linkedin/drelephant/spark/fetchers/SparkRestClientTest.scala index c20223fb8..00581a68b 100644 --- a/test/com/linkedin/drelephant/spark/fetchers/SparkRestClientTest.scala +++ b/test/com/linkedin/drelephant/spark/fetchers/SparkRestClientTest.scala @@ -300,7 +300,7 @@ object SparkRestClientTest { @Path("applications/{appId}/{attemptId}/jobs") def getJobs(): JobsResource = new JobsResource() - @Path("applications/{appId}/{attemptId}/stages") + @Path("applications/{appId}/{attemptId}/stages/withSummaries") def getStages(): StagesResource = new StagesResource() @Path("applications/{appId}/{attemptId}/allexecutors") @@ -382,7 +382,7 @@ object SparkRestClientTest { @Path("applications/{appId}/jobs") def getJobs(): JobsResource = new JobsResource() - @Path("applications/{appId}/stages") + @Path("applications/{appId}/stages/withSummaries") def getStages(): StagesResource = new StagesResource() @Path("applications/{appId}/allexecutors") diff --git a/test/com/linkedin/drelephant/spark/heuristics/ConfigurationParametersHeuristicTest.scala b/test/com/linkedin/drelephant/spark/heuristics/ConfigurationParametersHeuristicTest.scala new file mode 100644 index 000000000..38315d060 --- /dev/null +++ b/test/com/linkedin/drelephant/spark/heuristics/ConfigurationParametersHeuristicTest.scala @@ -0,0 +1,686 @@ +/* + * Copyright 2016 LinkedIn Corp. + * + * Licensed 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 com.linkedin.drelephant.spark.heuristics + +import com.linkedin.drelephant.analysis.{HeuristicResult, Severity} +import com.linkedin.drelephant.spark.heuristics.SparkTestUtilities._ +import com.linkedin.drelephant.spark.heuristics.ConfigurationHeuristicsConstants._ +import org.scalatest.{FunSpec, Matchers} + +class ConfigurationParametersHeuristicTest extends FunSpec with Matchers { + describe(".apply") { + it("unused executor memory, increase executor cores") { + // executor and driver memory are both over provisioned, increase executor + // cores and decrease driver memory + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).create(), + StageBuilder(2, 200).create(), + StageBuilder(3, 200).create(), + StageBuilder(4, 200).create(), + StageBuilder(5, 10).create() + ) + val executors = Seq( + createExecutorSummary("driver", 200, 3, 300), + createExecutorSummary("1", 800, 3, 300), + createExecutorSummary("2", 500, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "4G", + SPARK_DRIVER_MEMORY -> "6G", + SPARK_EXECUTOR_MEMORY_OVERHEAD -> "1G", + SPARK_DRIVER_MEMORY_OVERHEAD -> "2G", + SPARK_EXECUTOR_CORES -> "1", + SPARK_DRIVER_CORES -> "1", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.1" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedDetails = Map( + "Current spark.executor.memory" -> "4GB", + "Current spark.driver.memory" -> "6GB", + "Current spark.executor.cores" -> "1", + "Current spark.driver.cores" -> "1", + "Current spark.memory.fraction" -> "0.1", + "Current spark.executor.instances" -> "200", + "Current spark.yarn.executor.memoryOverhead" -> "1024MB", + "Current spark.yarn.driver.memoryOverhead" -> "2GB", + "Recommended spark.executor.cores" -> "3", + "Recommended spark.executor.memory" -> "4GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "1", + "Recommended spark.driver.memory" -> "640MB", + "Recommended spark.yarn.executor.memoryOverhead" -> "1024MB", + "Recommended spark.yarn.driver.memoryOverhead" -> "2GB", + "Recommended spark.executor.instances" -> "66") + checkHeuristicResults(result, Severity.CRITICAL, 16, expectedDetails) + } + + it("unused executor memory, decrease executor memory") { + // executor memory and driver memory are both over provisioned, decrease both + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).create(), + StageBuilder(2, 200).create(), + StageBuilder(3, 200).create(), + StageBuilder(4, 200).create(), + StageBuilder(5, 10).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1200, 3, 300), + createExecutorSummary("1", 800, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "4G", + SPARK_DRIVER_MEMORY -> "6G", + SPARK_EXECUTOR_MEMORY_OVERHEAD -> "1G", + SPARK_DRIVER_MEMORY_OVERHEAD -> "2G", + SPARK_EXECUTOR_CORES -> "5", + SPARK_DRIVER_CORES -> "5", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.1" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedDetails = Map( + "Current spark.executor.memory" -> "4GB", + "Current spark.driver.memory" -> "6GB", + "Current spark.executor.cores" -> "5", + "Current spark.driver.cores" -> "5", + "Current spark.memory.fraction" -> "0.1", + "Current spark.executor.instances" -> "200", + "Current spark.yarn.executor.memoryOverhead" -> "1024MB", + "Current spark.yarn.driver.memoryOverhead" -> "2GB", + "Recommended spark.executor.cores" -> "5", + "Recommended spark.executor.memory" -> "1550MB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "1800MB", + "Recommended spark.yarn.executor.memoryOverhead" -> "1024MB", + "Recommended spark.yarn.driver.memoryOverhead" -> "2GB", + "Recommended spark.executor.instances" -> "40") + checkHeuristicResults(result, Severity.CRITICAL, 17, expectedDetails) + } + + it("OOM, double memory") { + // tasks failed with OOM, since executor memory is low (2G), double the memory + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 60, 15000).create(), + StageBuilder(2, 20).taskRuntime(60, 60, 1200).failures(5, 5, 0).create(), + StageBuilder(3, 10).taskRuntime(100, 100, 1000).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "20", + SPARK_SQL_SHUFFLE_PARTITIONS -> "20", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 2 has 5 failed tasks.\n" + + "Stage 2 has 5 tasks that failed because of OutOfMemory exception." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "20", + "Recommended spark.executor.cores" -> "4", + "Recommended spark.executor.memory" -> "4GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "5", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 20, expectedDetails) + } + + it("OOM, increase memory by half") { + // tasks failed with OOM, since executor memory is moderate (6G), increase memory by .5 + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 60, 15000).create(), + StageBuilder(2, 20).taskRuntime(60, 60, 1200).failures(5, 5, 0).create(), + StageBuilder(3, 10).taskRuntime(100, 100, 1000).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 5000, 3, 300), + createExecutorSummary("2", 5500, 3, 300), + createExecutorSummary("3", 3000, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "6G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "20", + SPARK_SQL_SHUFFLE_PARTITIONS -> "20", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 2 has 5 failed tasks.\n" + + "Stage 2 has 5 tasks that failed because of OutOfMemory exception." + val expectedDetails = Map( + "Current spark.executor.memory" -> "6GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "20", + "Recommended spark.executor.cores" -> "4", + "Recommended spark.executor.memory" -> "9GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "5", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 20, expectedDetails) + } + + it("OOM, non-default partitions, decrease cores") { + // tasks failed with OOM, since executor memory is higher (9G), decrease cores + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 60, 15000).create(), + StageBuilder(2, 20).taskRuntime(60, 60, 1200).create(), + StageBuilder(3, 10).taskRuntime(100, 100, 1000).failures(5, 5, 0).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 8000, 3, 300), + createExecutorSummary("2", 6000, 3, 300), + createExecutorSummary("3", 6000, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "9G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "20", + SPARK_SQL_SHUFFLE_PARTITIONS -> "20", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 3 has 5 failed tasks.\n" + + "Stage 3 has 5 tasks that failed because of OutOfMemory exception." + val expectedDetails = Map( + "Current spark.executor.memory" -> "9GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "20", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "9GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "10", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 20, expectedDetails) + } + + it("OOM, non-default partitions, increase memory by 0.25") { + // tasks failed with OOM, since executor memory is higher (9G), increase memory 25% + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 60, 15000).create(), + StageBuilder(2, 20).taskRuntime(60, 60, 1200).create(), + StageBuilder(3, 10).taskRuntime(100, 100, 1000).failures(5, 5, 0).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 8000, 3, 300), + createExecutorSummary("2", 6000, 3, 300), + createExecutorSummary("3", 6000, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "9G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "2", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "20", + SPARK_SQL_SHUFFLE_PARTITIONS -> "20", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 3 has 5 failed tasks.\n" + + "Stage 3 has 5 tasks that failed because of OutOfMemory exception." + val expectedDetails = Map( + "Current spark.executor.memory" -> "9GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "2", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "20", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "12GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "10", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 20, expectedDetails) + } + + it("Container killed by yarn") { + // tasks failed with container killed by YARN, increase memory overhead + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 60, 15000).create(), + StageBuilder(2, 20).taskRuntime(60, 60, 1200).create(), + StageBuilder(3, 10).taskRuntime(100, 100, 1000).failures(7, 0, 5).create() + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 8000, 3, 300), + createExecutorSummary("2", 6000, 3, 300), + createExecutorSummary("3", 6000, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "9G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "2", + SPARK_DRIVER_CORES -> "2", + SPARK_SQL_SHUFFLE_PARTITIONS -> "20", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 3 has 7 failed tasks.\n" + + "Stage 3 has 5 tasks that failed because the container was killed by YARN for exceeding memory limits." + val expectedDetails = Map( + "Current spark.executor.memory" -> "9GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "2", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "9GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.yarn.executor.memoryOverhead" -> "1946MB", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 28, expectedDetails) + } + + it("long tasks and execution memory spill 1") { + // There are both long tasks and execution memory spill, increasing the number of + // partitions for long tasks to 800 is sufficient to fix execution memory spill as well + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 50, 1500).create(), + StageBuilder(2, 200).taskRuntime(600, 600, 120000).shuffleRead(500, 600, 12000).spill(50, 60, 1200).create(), + StageBuilder(3, 200).taskRuntime(30, 80, 1200).create(), + StageBuilder(4, 10).taskRuntime(100, 100, 1000).create()) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 2 has a long median task run time of 10.00 min.\n" + + "Stage 2 has 200 tasks, 0 B input, 11.72 GB shuffle read, 0 B shuffle write, and 0 B output.\n" + + "Stage 2 has 1.17 GB execution memory spill." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "200", + "Recommended spark.executor.cores" -> "4", + "Recommended spark.executor.memory" -> "4GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "50", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.SEVERE, 1000, expectedDetails) + } + + it("long tasks and execution memory spill 2") { + // There are both long tasks and execution memory spill, increase the number of + // partitions and executor memory. + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 50, 1500).create(), + StageBuilder(2, 200).taskRuntime(90, 90, 1800).shuffleRead(500, 600, 12000).spill(500, 600, 600000).create(), + StageBuilder(3, 200).taskRuntime(900, 900, 180000).shuffleRead(500, 600, 12000).spill(50, 60, 1200).create(), + StageBuilder(4, 10).taskRuntime(100, 100, 1000).create()) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 2 has 585.94 GB execution memory spill.\n" + + "Stage 3 has a long median task run time of 15.00 min.\n" + + "Stage 3 has 200 tasks, 0 B input, 11.72 GB shuffle read, 0 B shuffle write, and 0 B output.\n" + + "Stage 3 has 1.17 GB execution memory spill." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "200", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "10GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "100", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 2000, expectedDetails) + } + + it("long tasks and execution memory spill 3") { + // There are both long tasks and execution memory spill, increase partitions + // and memory, and decrease cores. + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 50, 1500).create(), + StageBuilder(2, 200).taskRuntime(60, 60, 1200).shuffleRead(500, 600, 12000).spill(500, 600, 1200000).create(), + StageBuilder(3, 200).taskRuntime(600, 600, 120000).shuffleRead(500, 600, 12000).spill(50, 60, 1200).create(), + StageBuilder(4, 10).taskRuntime(100, 100, 1000).create()) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 2 has 1.14 TB execution memory spill.\n" + + "Stage 3 has a long median task run time of 10.00 min.\n" + + "Stage 3 has 200 tasks, 0 B input, 11.72 GB shuffle read, 0 B shuffle write, and 0 B output.\n" + + "Stage 3 has 1.17 GB execution memory spill." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "200", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "10GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "100", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 1800, expectedDetails) + } + + it("execution memory spill 1") { + // Execution memory spill for stages 2 and 3, stage 2 is processing a lot of data. + // Values should be calculated for stage 2, but do not flag stage 2. + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 50, 1500).create(), + StageBuilder(2, 200).taskRuntime(60, 60, 1200).shuffleRead(50000, 60000, 4000000).spill(500, 600, 4000000).create(), + StageBuilder(3, 200).taskRuntime(60, 60, 1200).shuffleRead(500, 600, 12000).spill(50, 60, 1200).create(), + StageBuilder(4, 10).taskRuntime(100, 100, 1000).create() + + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = + "Stage 2: a large amount of data is being processesd. Examine the application to see if this can be reduced.\n" + + "Stage 2 has 3.81 TB execution memory spill.\n" + + "Stage 2 has 200 tasks, 0 B input read, 3.81 TB shuffle read, 0 B shuffle write, 0 B output.\n" + + "Stage 2 has median task values: 500 MB memory spill, 0 B input, 48.83 GB shuffle read, 0 B shuffle write, 0 B output.\n" + + "Stage 3 has 1.17 GB execution memory spill." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "200", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "10GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "100", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.MODERATE, 400, expectedDetails) + } + + it("execution memory spill 2") { + // Execution memory spill for stages 1 and 2, where stage 1 has different number of partitions. + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(50, 50, 1500).input(50000, 60000, 2000000).spill(5000, 6000, 2000000).create(), + StageBuilder(2, 200).taskRuntime(60, 60, 1200).shuffleRead(5000, 6000, 400000).spill(500, 600, 400000).create(), + StageBuilder(3, 200).taskRuntime(60, 60, 1200).create(), + StageBuilder(4, 10).taskRuntime(100, 100, 1000).create() + + ) + val executors = Seq( + createExecutorSummary("driver", 1500, 3, 300), + createExecutorSummary("1", 1500, 3, 300), + createExecutorSummary("2", 1000, 3, 300), + createExecutorSummary("3", 800, 3, 300) + ) + val properties = Map( + SPARK_EXECUTOR_MEMORY -> "2G", + SPARK_DRIVER_MEMORY -> "2G", + SPARK_EXECUTOR_CORES -> "4", + SPARK_DRIVER_CORES -> "2", + SPARK_EXECUTOR_INSTANCES -> "200", + SPARK_SQL_SHUFFLE_PARTITIONS -> "200", + SPARK_MEMORY_FRACTION -> "0.6" + ) + + val data = createSparkApplicationData(stages, executors, Some(properties)) + + val configurationParametersHeuristic = new ConfigurationParametersHeuristic( + heuristicConfigurationData) + val evaluator = new ConfigurationParametersHeuristic.Evaluator( + configurationParametersHeuristic, data) + + val result = configurationParametersHeuristic.apply(data) + val expectedStageDetails = "Stage 1 has 1.91 TB execution memory spill.\n" + + "Stage 2 has 390.62 GB execution memory spill." + val expectedDetails = Map( + "Current spark.executor.memory" -> "2GB", + "Current spark.driver.memory" -> "2GB", + "Current spark.executor.cores" -> "4", + "Current spark.driver.cores" -> "2", + "Current spark.memory.fraction" -> "0.6", + "Current spark.executor.instances" -> "200", + "Recommended spark.executor.cores" -> "2", + "Recommended spark.executor.memory" -> "10GB", + "Recommended spark.memory.fraction" -> "0.6", + "Recommended spark.driver.cores" -> "2", + "Recommended spark.driver.memory" -> "2GB", + "Recommended spark.executor.instances" -> "100", + "stage details" -> expectedStageDetails) + checkHeuristicResults(result, Severity.CRITICAL, 812, expectedDetails) + } + } + + /** + * Check if the calculated heuristic results match the expected values. + * + * @param actual the actual heuristic result. + * @param expectedSeverity expected severity. + * @param expectedScore expeced score. + * @param expectedDetails expected details. + */ + private def checkHeuristicResults( + actual: HeuristicResult, + expectedSeverity: Severity, + expectedScore: Int, + expectedDetails: Map[String, String]) = { + actual.getSeverity should be(expectedSeverity) + actual.getScore should be(expectedScore) + actual.getHeuristicResultDetails.size() should be(expectedDetails.size) + (0 until expectedDetails.size).foreach { i => + val actualDetail = actual.getHeuristicResultDetails.get(i) + Some(actualDetail.getValue) should be (expectedDetails.get(actualDetail.getName)) + } + } +} diff --git a/test/com/linkedin/drelephant/spark/heuristics/SparkTestUtilities.scala b/test/com/linkedin/drelephant/spark/heuristics/SparkTestUtilities.scala new file mode 100644 index 000000000..e284135be --- /dev/null +++ b/test/com/linkedin/drelephant/spark/heuristics/SparkTestUtilities.scala @@ -0,0 +1,557 @@ +/* + * Copyright 2016 LinkedIn Corp. + * + * Licensed 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 com.linkedin.drelephant.spark.heuristics + +import java.util.Date + +import com.linkedin.drelephant.analysis.{ApplicationType, Severity} +import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData +import com.linkedin.drelephant.spark.data.{SparkApplicationData, SparkLogDerivedData, SparkRestDerivedData} +import com.linkedin.drelephant.spark.fetchers.statusapiv1._ +import org.apache.spark.scheduler.SparkListenerEnvironmentUpdate + +import scala.collection.JavaConverters + +private [heuristics] object SparkTestUtilities { + import JavaConverters._ + import java.text.SimpleDateFormat + + val OOM_ERROR = "java.lang.OutOfMemoryError" + val OVERHEAD_MEMORY_ERROR = "killed by YARN for exceeding memory limits" + + private val sdf = new SimpleDateFormat("dd/MM/yyyy HH:mm:ss") + + /** Create a ample heuristics configuration data. */ + def createHeuristicConfigurationData( + params: Map[String, String] = Map.empty): HeuristicConfigurationData = + new HeuristicConfigurationData("heuristic", "class", "view", new ApplicationType("type"), params.asJava) + + /** + * Builder for creating a StageAnalysis. + * + * @param stageId stage ID. + * @param numTasks total number of tasks for the stage. + */ + case class StageAnalysisBuilder(stageId: Int, numTasks: Int) { + var rawSpillSeverity = Severity.NONE + var executionSpillSeverity = Severity.NONE + var longTaskSeverity = Severity.NONE + var rawSkewSeverity = Severity.NONE + var taskSkewSeverity = Severity.NONE + var failedWithOOMSeverity = Severity.NONE + var failedWithContainerKilledSeverity = Severity.NONE + var gcSeverity = Severity.NONE + var taskFailureSeverity = Severity.NONE + var stageFailureSeverity = Severity.NONE + var spillScore = 0 + var longTaskScore = 0 + var taskSkewScore = 0 + var taskFailureScore = 0 + var stageFailureScore = 0 + var gcScore = 0 + var medianRunTime: Option[Double] = None + var maxRunTime: Option[Double] = None + var memoryBytesSpilled = 0L + var maxTaskBytesSpilled = 0L + var inputBytes: Long = 0L + var outputBytes: Long = 0L + var shuffleReadBytes: Long = 0L + var shuffleWriteBytes: Long = 0L + var numFailedTasks = 0 + var numTasksWithOOM = 0 + var numTasksWithContainerKilled = 0 + var stageDuration = Some((5 * 60 * 1000).toLong) + var spillDetails: Seq[String] = Seq() + var longTaskDetails: Seq[String] = Seq() + var taskSkewDetails: Seq[String] = Seq() + var taskFailureDetails: Seq[String] = Seq() + var stageFailureDetails: Seq[String] = Seq() + var gcDetails: Seq[String] = Seq() + + /** + * Configure execution memory spill related parameters. + * + * @param raw the raw execution memory spill severity. + * @param severity the reported execution memory spill severity. + * @param maxTaskSpillMb maximum amount (MB) of execution memory spill for a task. + * @param bytesSpilledMb total amount (MB) of execution memory spill. + * @return this StageAnalysisBuilder. + */ + def spill( + raw: Severity, + severity: Severity, + score: Int, + maxTaskSpillMb: Long, + bytesSpilledMb: Long, + details: Seq[String]): StageAnalysisBuilder = { + rawSpillSeverity = raw + executionSpillSeverity = severity + spillScore = score + maxTaskBytesSpilled = maxTaskSpillMb << 20 + memoryBytesSpilled = bytesSpilledMb << 20 + spillDetails = details + this + } + + /** Set the amount of input data in MB. */ + def input(inputMb: Long): StageAnalysisBuilder = { + inputBytes = inputMb << 20 + this + } + + /** Set the amount of output data in MB. */ + def output(outputMb: Long): StageAnalysisBuilder = { + outputBytes = outputMb << 20 + this + } + + /** Set the amount of shuffle read data in MB. */ + def shuffleRead(shuffleReadMb: Long): StageAnalysisBuilder = { + shuffleReadBytes = shuffleReadMb << 20 + this + } + + /** Set the amount of shuffle write data in MB. */ + def shuffleWrite(shuffleWriteMb: Long): StageAnalysisBuilder = { + shuffleWriteBytes = shuffleWriteMb << 20 + this + } + + /** Set the stage duration. */ + def duration(sec: Long): StageAnalysisBuilder = { + stageDuration = Some(sec * 1000) + this + } + + /** Set the median and max task runtimes in seconds */ + def taskRuntime(median: Double, maximum: Double): StageAnalysisBuilder = { + medianRunTime = Some(median * 1000) + maxRunTime = Some(maximum * 1000) + this + } + + /** set the long task analysis information */ + def longTask(severity: Severity, score: Int, details: Seq[String]): StageAnalysisBuilder = { + longTaskSeverity = severity + longTaskScore = score + longTaskDetails = details + this + } + + /** set the raw and reported task skew severity and details */ + def skew( + raw: Severity, + severity: Severity, + score: Int, + details: Seq[String]): StageAnalysisBuilder = { + rawSkewSeverity = raw + taskSkewSeverity = severity + taskSkewScore = score + taskSkewDetails = details + this + } + + /** + * Configure stage failure information. + * + * @param severity severity of stage failure. + * @param score score for stage failure analysis + * @param details information and recommendations + * @return + */ + def stageFailure(severity: Severity, + score: Int, + details: Seq[String]): StageAnalysisBuilder = { + stageFailureSeverity = severity + stageFailureScore = score + stageFailureDetails = details + this + } + + /** + * Configure task failure information. + * + * @param taskSeverity severity of all task failures. + * @param oomSeverity severity of task failures due to OutOfMemory errors. + * @param containerKilledSeverity severity of failures due to containers killed by YARN. + * @param score score from task failure analysis. + * @param numFailures total number of task failures. + * @param numOOM total number of tasks failed with OutOfMemory errors. + * @param numContainerKilled total number of tasks failed due to container killed by YARN. + * @param details information and recommendations for task failures + * @return this StageAnalysisBuilder. + */ + def taskFailures( + taskSeverity: Severity, + oomSeverity: Severity, + containerKilledSeverity: Severity, + score: Int, + numFailures: Int, + numOOM: Int, + numContainerKilled: Int, + details: Seq[String]): StageAnalysisBuilder = { + taskFailureSeverity = taskSeverity + failedWithOOMSeverity = oomSeverity + failedWithContainerKilledSeverity = containerKilledSeverity + taskFailureScore = score + numFailedTasks = numFailures + numTasksWithOOM = numOOM + numTasksWithContainerKilled = numContainerKilled + taskFailureDetails = details + this + } + + /** Create the StageAnalysis. */ + def create(): StageAnalysis = { + StageAnalysis( + stageId, + ExecutionMemorySpillResult(executionSpillSeverity, spillScore, spillDetails, + rawSpillSeverity, memoryBytesSpilled, maxTaskBytesSpilled), + SimpleStageAnalysisResult(longTaskSeverity, longTaskScore, longTaskDetails), + TaskSkewResult(taskSkewSeverity, taskSkewScore, taskSkewDetails, rawSkewSeverity), + TaskFailureResult(taskFailureSeverity, taskFailureScore, taskFailureDetails, + failedWithOOMSeverity, failedWithContainerKilledSeverity, numFailedTasks, + numTasksWithOOM, numTasksWithContainerKilled), + SimpleStageAnalysisResult(stageFailureSeverity, stageFailureScore, stageFailureDetails), + numTasks, medianRunTime, maxRunTime, stageDuration, inputBytes, outputBytes, + shuffleReadBytes, shuffleWriteBytes) + } + } + + /** + * Builder for creating StageData. + * + * @param stageId stage ID + * @param numTasks total number of tasks for the stage. + */ + case class StageBuilder(stageId: Int, numTasks: Int) { + val stage = new StageDataImpl( + StageStatus.COMPLETE, + stageId, + attemptId = 0, + numTasks = numTasks, + numActiveTasks = numTasks, + numCompleteTasks = numTasks, + numFailedTasks = 0, + executorRunTime = 0, + executorCpuTime = 0, + submissionTime = Some(sdf.parse("09/09/2018 12:00:00")), + firstTaskLaunchedTime = None, + completionTime = Some(sdf.parse("09/09/2018 12:05:00")), + failureReason = None, + + inputBytes = 0, + inputRecords = 0, + outputBytes = 0, + outputRecords = 0, + shuffleReadBytes = 0, + shuffleReadRecords = 0, + shuffleWriteBytes = 0, + shuffleWriteRecords = 0, + memoryBytesSpilled = 0, + diskBytesSpilled = 0, + name = "foo", + details = "stage details", + schedulingPool = "", + accumulatorUpdates = Seq.empty, + tasks = None, + executorSummary = None, + peakJvmUsedMemory = None, + peakExecutionMemory = None, + peakStorageMemory = None, + peakUnifiedMemory = None, + taskSummary = None, + executorMetricsSummary = None + ) + + /** Create the specified number of tasks for the stage. */ + private def createTasks(numTasks: Int): Map[Long, TaskDataImpl] = { + (1 until (numTasks + 1)).map { i => + (i.toLong, new TaskDataImpl( + taskId = i.toLong, + index = 1, + attempt = 0, + launchTime = new Date(), + executorId = "1", + host = "SomeHost", + taskLocality = "ANY", + speculative = false, + accumulatorUpdates = Seq(), + errorMessage = None, + taskMetrics = None)) + }.toMap + } + + /** Set the stage status. */ + def status(stageStatus: StageStatus, reason: Option[String]): StageBuilder = { + stage.status = stageStatus + stage.failureReason = reason + this + } + + /** + * Set the run times. + * + * @param medianS median task run time in seconds. + * @param maxS maximum task runtime in seconds. + * @param totalS total runtime for all tasks. + * @return this StageBuilder. + */ + def taskRuntime(medianS: Int, maxS: Int, totalS: Int): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianMs = (medianS * 1000).toDouble + val maxMs = (maxS * 1000).toDouble + taskMetricDistributions.executorRunTime = + IndexedSeq(medianMs/2, medianMs, medianMs, medianMs, maxMs) + stage.executorRunTime = totalS * 1000 + this + } + + /** + * Set the input information. + * + * @param medianMB median amount of input read for a task in MB. + * @param maxMB maximum amount of input read for a task in MB. + * @param totalMB total amount of input read for the stage in MB. + * @return this StageBuilder. + */ + def input(medianMB: Long, maxMB: Long, totalMB: Long): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianBytes = (medianMB << 20).toDouble + val maxBytes = (maxMB << 20).toDouble + taskMetricDistributions.inputMetrics = + Some(new InputMetricDistributionsImpl( + IndexedSeq(medianBytes/2, medianBytes, medianBytes, medianBytes, maxBytes), + IndexedSeq(1000.0, 2000.0, 2000.0, 2000.0, 3000.0))) + stage.inputBytes = totalMB << 20 + this + } + + /** + * Set the output information. + * + * @param medianMB median amount of output written for a task in MB. + * @param maxMB maximum amount of output written for a task in MB. + * @param totalMB total amount of output written for the stage in MB. + * @return this StageBuilder. + */ + def output(medianMB: Long, maxMB: Long, totalMB: Long): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianBytes = (medianMB << 20).toDouble + val maxBytes = (maxMB << 20).toDouble + taskMetricDistributions.outputMetrics = + Some(new OutputMetricDistributionsImpl( + IndexedSeq(medianBytes/2, medianBytes, medianBytes, medianBytes, maxBytes), + IndexedSeq(1000.0, 2000.0, 2000.0, 2000.0, 3000.0))) + stage.outputBytes = totalMB << 20 + this + } + + /** + * Set the shuffle read information. + * + * @param medianMB median amount of shuffle read for a task in MB. + * @param maxMB maximum amount of shuffle read for a task in MB. + * @param totalMB total amount of shuffle read for the stage in MB. + * @return this StageBuilder. + */ + def shuffleRead(medianMB: Long, maxMB: Long, totalMB: Long): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianBytes = (medianMB << 20).toDouble + val maxBytes = (maxMB << 20).toDouble + taskMetricDistributions.shuffleReadMetrics = + Some(new ShuffleReadMetricDistributionsImpl( + IndexedSeq(medianBytes/2, medianBytes, medianBytes, medianBytes, maxBytes), + IndexedSeq(1000.0, 2000.0, 2000.0, 2000.0, 3000.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0))) + stage.shuffleReadBytes = totalMB << 20 + this + } + + /** + * Set the shuffle write information. + * + * @param medianMB median amount of shuffle write for a task in MB. + * @param maxMB maximum amount of shuffle write for a task in MB. + * @param totalMB total amount of shuffle write for the stage in MB. + * @return this StageBuilder. + */ + def shuffleWrite(medianMB: Long, maxMB: Long, totalMB: Long): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianBytes = (medianMB << 20).toDouble + val maxBytes = (maxMB << 20).toDouble + taskMetricDistributions.shuffleWriteMetrics = + Some(new ShuffleWriteMetricDistributionsImpl( + IndexedSeq(medianBytes/2, medianBytes, medianBytes, medianBytes, maxBytes), + IndexedSeq(1000.0, 2000.0, 2000.0, 2000.0, 3000.0), + IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0))) + stage.shuffleWriteBytes = totalMB << 20 + this + } + + /** + * Set the execution memory spill information. + * + * @param medianMB median amount of execution memory spill for a task in MB. + * @param maxMB maximum amount of iexecution memory spill for a task in MB. + * @param totalMB total amount of execution memory spill for the stage in MB. + * @return this StageBuilder. + */ + def spill(medianMB: Long, maxMB: Long, totalMB: Long): StageBuilder = { + val taskMetricDistributions = getTaskMetricDistributions() + val medianBytes = (medianMB << 20).toDouble + val maxBytes = (maxMB << 20).toDouble + taskMetricDistributions.memoryBytesSpilled = + IndexedSeq(medianBytes/2, medianBytes, medianBytes, medianBytes, maxBytes) + stage.memoryBytesSpilled = totalMB << 20 + this + } + + /** + * Set the failure information. + * + * @param numFailed total number of tasks failed. + * @param numOOM total number of tasks which failed due to OutOfMemory. + * @param numContainerKilled total number of ask which failed due to container killed by YARN. + * @return this StageBuilder. + */ + def failures(numFailed: Int, numOOM: Int, numContainerKilled: Int): StageBuilder = { + stage.tasks = Some(createTasks(numFailed)) + (1 until (numOOM + 1)).map { i => + stage.tasks.get(i.toLong).errorMessage = Some(OOM_ERROR) + } + ((numOOM + 1) until (numOOM + numContainerKilled + 1)).map { i => + stage.tasks.get(i.toLong).errorMessage = Some(OVERHEAD_MEMORY_ERROR) + } + ((numOOM + numContainerKilled + 1) until numFailed + 1).map { i => + stage.tasks.get(i.toLong).errorMessage = Some("ArrayIndexOutOfBoundsException") + } + stage.numFailedTasks = numFailed + this + } + + /** Set the stage submission and completion times. */ + def times(submissionTime: String, completionTime: String): StageBuilder = { + stage.submissionTime = Some(sdf.parse(submissionTime)) + stage.completionTime = Some(sdf.parse(completionTime)) + this + } + + /** Create the StageDataImpl. */ + def create(): StageDataImpl = stage + + /** @return a askMetricDistributionsImpl for the StageData, creating it if needed. */ + private def getTaskMetricDistributions(): TaskMetricDistributionsImpl = { + stage.taskSummary match { + case None => + val taskMetricDistributions = + new TaskMetricDistributionsImpl( + quantiles = IndexedSeq(0.0, 0.25, 0.5, 0.75, 1.0), + executorDeserializeTime = IndexedSeq(0.0, 0.0, 0.1, 0.1, 0.2), + executorDeserializeCpuTime = IndexedSeq(0.0, 0.0, 0.1, 0.1, 0.2), + executorRunTime = IndexedSeq(1000.0, 5000.0, 6000.0, 6500.0, 7000.0), + executorCpuTime = IndexedSeq(1000.0, 5000.0, 6000.0, 6500.0, 7000.0), + resultSize = IndexedSeq(0.0, 0.0, 0.0, 0.0), + jvmGcTime = IndexedSeq(0.0, 0.0, 0.0, 0.0), + resultSerializationTime = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + gettingResultTime = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + schedulerDelay = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + peakExecutionMemory = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + memoryBytesSpilled = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + diskBytesSpilled = IndexedSeq(0.0, 0.0, 0.0, 0.0, 0.0), + inputMetrics = None, + outputMetrics = None, + shuffleReadMetrics = None, + shuffleWriteMetrics = None) + stage.taskSummary = Some(taskMetricDistributions) + taskMetricDistributions + case Some(taskMetricDistributions) => + taskMetricDistributions + } + } + } + + /** + * Create an executor metrics summary. + * + * @param id executor ID + * @param jvmUsedMemoryMb peak JVM used memory for the executor. + * @param totalGCTimeSec total time spent in GC by the executor. + * @param totalDurationSec total task runtime for the executor. + * @return executor summary. + */ + private[heuristics] def createExecutorSummary( + id: String, + jvmUsedMemoryMb: Long, + totalGCTimeSec: Long, + totalDurationSec: Long): ExecutorSummaryImpl = new ExecutorSummaryImpl( + id, + hostPort = "", + rddBlocks = 0, + memoryUsed=0, + diskUsed = 0, + activeTasks = 0, + failedTasks = 0, + completedTasks = 0, + totalTasks = 0, + maxTasks = 0, + totalDurationSec * 1000, + totalInputBytes=0, + totalShuffleRead=0, + totalShuffleWrite= 0, + maxMemory = 0, + totalGCTimeSec * 1000, + totalMemoryBytesSpilled = 0, + executorLogs = Map.empty, + peakJvmUsedMemory = Map("jvmUsedMemory" -> (jvmUsedMemoryMb << 20)), + peakUnifiedMemory = Map.empty + ) + + /** + * Create the Spark application data. + * + * @param stages list of stage data + * @param executorSummaries list of executor summaries. + * @param properties configuration properties for the Spark application. + * @return Spark application data. + */ + def createSparkApplicationData + (stages: Seq[StageDataImpl], + executorSummaries: Seq[ExecutorSummaryImpl], + properties: Option[Map[String, String]]): SparkApplicationData = { + val appId = "application_1" + + val logDerivedData = properties.map { props => + SparkLogDerivedData( + SparkListenerEnvironmentUpdate(Map("Spark Properties" -> props.toSeq)) + )} + + val restDerivedData = SparkRestDerivedData( + new ApplicationInfoImpl(appId, name = "app", Seq.empty), + jobDatas = Seq.empty, + stageDatas = stages, + executorSummaries = executorSummaries, + stagesWithFailedTasks = stages + ) + SparkApplicationData(appId, restDerivedData, logDerivedData) + } +} diff --git a/test/com/linkedin/drelephant/spark/heuristics/StagesAnalyzerTest.scala b/test/com/linkedin/drelephant/spark/heuristics/StagesAnalyzerTest.scala new file mode 100644 index 000000000..3e814900e --- /dev/null +++ b/test/com/linkedin/drelephant/spark/heuristics/StagesAnalyzerTest.scala @@ -0,0 +1,480 @@ +/* + * Copyright 2016 LinkedIn Corp. + * + * Licensed 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 com.linkedin.drelephant.spark.heuristics + +import java.util.Date + +import com.linkedin.drelephant.analysis.{ApplicationType, Severity} +import com.linkedin.drelephant.spark.fetchers.statusapiv1._ +import org.scalatest.{FunSpec, Matchers} + +import scala.collection.JavaConverters + +/** Tests for the StagesAnalyzer. */ +class StagesAnalyzerTest extends FunSpec with Matchers { + import SparkTestUtilities._ + + describe("StagesAnalyzer") { + it("has task failures severity") { + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).create(), + StageBuilder(2, 5).failures(2, 2, 0).create(), + StageBuilder(3, 15).failures(2, 0, 1).create(), + StageBuilder(4, 15).failures(3, 1, 2).create(), + StageBuilder(5, 4).failures(2, 0, 0).status(StageStatus.FAILED, Some("array issues")).create()) + val properties = Map( "spark.sql.shuffle.partitions" -> "200") + val data = createSparkApplicationData(stages, Seq.empty, Some(properties)) + + val expectedAnalysis = Seq( + StageAnalysisBuilder(1, 3).create(), + StageAnalysisBuilder(2, 5) + .taskFailures(Severity.CRITICAL, Severity.CRITICAL, Severity.NONE, 8, 2, 2, 0, + Seq("Stage 2 has 2 failed tasks.", + "Stage 2 has 2 tasks that failed because of OutOfMemory exception.")) + .create(), + StageAnalysisBuilder(3, 15) + .taskFailures(Severity.MODERATE, Severity.NONE, Severity.LOW, 4, 2, 0, 1, + Seq("Stage 3 has 2 failed tasks.", + "Stage 3 has 1 tasks that failed because the container was killed by YARN for exceeding memory limits.")) + .create(), + StageAnalysisBuilder(4, 15) + .taskFailures(Severity.CRITICAL, Severity.LOW, Severity.MODERATE, 12, 3, 1, 2, + Seq("Stage 4 has 3 failed tasks.", + "Stage 4 has 1 tasks that failed because of OutOfMemory exception.", + "Stage 4 has 2 tasks that failed because the container was killed by YARN for exceeding memory limits.")) + .create(), + StageAnalysisBuilder(5, 4) + .taskFailures(Severity.CRITICAL, Severity.NONE, Severity.NONE, 8, 2, 0, 0, + Seq("Stage 5 has 2 failed tasks.")) + .stageFailure(Severity.CRITICAL, 16, Seq("Stage 5 failed: array issues")) + .create()) + + val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data) + val stageAnalysis = stageAnalyzer.getStageAnalysis() + (0 until expectedAnalysis.size).foreach { i => + compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i)) + } + } + + it("has task skew severity") { + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 5).taskRuntime(200, 250, 600).create(), + StageBuilder(2, 5).taskRuntime(100, 250, 260).input(5, 250, 260).create(), + StageBuilder(3, 5).taskRuntime(20, 250, 53).create(), + StageBuilder(4, 5).taskRuntime(5, 250, 260).input(5, 250, 260).create(), + StageBuilder(5, 5).taskRuntime(50, 250, 350).shuffleRead(50, 250, 350).shuffleWrite(50, 250, 400).create(), + StageBuilder(6, 5).taskRuntime(50, 250, 350).shuffleRead(50, 50, 50).output(50, 50, 50).create(), + StageBuilder(7, 5).taskRuntime(20, 250, 290).shuffleWrite(250, 250, 600).output(20, 250, 290).create(), + StageBuilder(8, 3).taskRuntime(200, 250, 1000).create(), + StageBuilder(9, 3).taskRuntime(5, 250, 70).create(), + StageBuilder(10, 3).taskRuntime(20, 250, 300).input(20, 250, 300).create(), + StageBuilder(11, 3).taskRuntime(50, 250, 350).shuffleRead(50, 250, 350).create(), + StageBuilder(12, 5).taskRuntime(2, 50, 53).times("09/09/2018 12:00:00", "09/09/2018 12:01:00").create(), + StageBuilder(13, 5).taskRuntime(5, 50, 60).input(50, 500, 600).create(), + StageBuilder(14, 5).taskRuntime(5, 200, 210).output(5, 200, 210).create()) + val properties = Map( "spark.sql.shuffle.partitions" -> "5") + val data = createSparkApplicationData(stages, Seq.empty, Some(properties)) + + val expectedAnalysis = Seq( + StageAnalysisBuilder(1, 5).taskRuntime(200, 250) + .longTask(Severity.LOW, 0, Seq()).create(), + StageAnalysisBuilder(2, 5).taskRuntime(100, 250).input(260) + .skew(Severity.LOW, Severity.LOW, 0, + Seq()).create(), + StageAnalysisBuilder(3, 5).taskRuntime(20, 250) + .skew(Severity.SEVERE, Severity.SEVERE, 15, + Seq("Stage 3 has skew in task run time (median is 20.00 sec, max is 4.17 min).", + "Stage 3: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(4, 5).taskRuntime(5, 250).input(260) + .skew(Severity.CRITICAL, Severity.CRITICAL, 20, + Seq("Stage 4 has skew in task run time (median is 5.00 sec, max is 4.17 min).", + "Stage 4 has skew in task input bytes (median is 5 MB, max is 250 MB).", + "Stage 4: please try to modify the application to make the input partitions more even.")).create(), + StageAnalysisBuilder(5, 5).taskRuntime(50, 250).shuffleRead(350).shuffleWrite(400) + .skew(Severity.MODERATE, Severity.MODERATE, 10, + Seq("Stage 5 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 5 has skew in task shuffle read bytes (median is 50 MB, max is 250 MB).", + "Stage 5 has skew in task shuffle write bytes (median is 50 MB, max is 250 MB).", + "Stage 5: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(6, 5).taskRuntime(50, 250).shuffleRead(50).output(50) + .skew(Severity.MODERATE, Severity.MODERATE, 10, + Seq( "Stage 6 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 6: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(7, 5).taskRuntime(20, 250).shuffleWrite(600).output(290) + .skew(Severity.SEVERE, Severity.SEVERE, 15, + Seq("Stage 7 has skew in task run time (median is 20.00 sec, max is 4.17 min).", + "Stage 7 has skew in task output bytes (median is 20 MB, max is 250 MB).", + "Stage 7: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(8, 3).taskRuntime(200, 250) + .longTask(Severity.LOW, 0, Seq()).create(), + StageAnalysisBuilder(9, 3).taskRuntime(5, 250) + .skew(Severity.CRITICAL, Severity.CRITICAL, 12, + Seq("Stage 9 has skew in task run time (median is 5.00 sec, max is 4.17 min).", + "Stage 9: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(10, 3).taskRuntime(20, 250).input(300) + .skew(Severity.SEVERE, Severity.SEVERE, 9, + Seq("Stage 10 has skew in task run time (median is 20.00 sec, max is 4.17 min).", + "Stage 10 has skew in task input bytes (median is 20 MB, max is 250 MB).", + "Stage 10: please try to modify the application to make the input partitions more even.")).create(), + StageAnalysisBuilder(11, 3).taskRuntime(50, 250).shuffleRead(350) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 11 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 11 has skew in task shuffle read bytes (median is 50 MB, max is 250 MB).", + "Stage 11: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(12, 5).taskRuntime(2, 50).duration(60) + .skew(Severity.CRITICAL, Severity.NONE, 0, + Seq()).create(), + StageAnalysisBuilder(13, 5).taskRuntime(5, 50).input(600) + .skew(Severity.SEVERE, Severity.NONE, 0, + Seq()).create(), + StageAnalysisBuilder(14, 5).taskRuntime(5, 200).output(210) + .skew(Severity.CRITICAL, Severity.NONE, 0, + Seq()).create()) + + val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data) + val stageAnalysis = stageAnalyzer.getStageAnalysis() + (0 until expectedAnalysis.size).foreach { i => + compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i)) + } + } + + it("has long task severity") { + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 3).taskRuntime(120, 150, 300).create(), + StageBuilder(2, 3).taskRuntime(180, 200, 400).create(), + StageBuilder(3, 3).taskRuntime(400, 500, 1000).create(), + StageBuilder(4, 3).taskRuntime(700, 900, 2000).create(), + StageBuilder(5, 3).taskRuntime(1200, 1500, 4000).create(), + StageBuilder(6, 3).taskRuntime(700, 3500, 4500).create(), + StageBuilder(7, 2).taskRuntime(700, 900, 2000).create(), + StageBuilder(8, 3).taskRuntime(3000, 3000, 9000).input(2 << 20, 3 << 20, 5 << 20).create(), + StageBuilder(9, 4003).taskRuntime(3000, 3000, 9000).shuffleRead(2 << 20, 3 << 20, 5 << 20).create(), + StageBuilder(10, 4000).taskRuntime(700, 900, 2000).create()) + val properties = Map( "spark.sql.shuffle.partitions" -> "3") + val data = createSparkApplicationData(stages, Seq.empty, Some(properties)) + + val expectedAnalysis = Seq( + StageAnalysisBuilder(1, 3).taskRuntime(120, 150).create(), + StageAnalysisBuilder(2, 3).taskRuntime(180, 200).longTask(Severity.LOW, 0, Seq()).create(), + StageAnalysisBuilder(3, 3).taskRuntime(400, 500).longTask(Severity.MODERATE, 6, + Seq("Stage 3 has a long median task run time of 6.67 min.", + "Stage 3 has 3 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.")).create(), + StageAnalysisBuilder(4, 3).taskRuntime(700, 900).longTask(Severity.SEVERE, 9, + Seq("Stage 4 has a long median task run time of 11.67 min.", + "Stage 4 has 3 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.")).create(), + StageAnalysisBuilder(5, 3).taskRuntime(1200, 1500).longTask(Severity.CRITICAL, 12, + Seq("Stage 5 has a long median task run time of 20.00 min.", + "Stage 5 has 3 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.")).create(), + StageAnalysisBuilder(6, 3).taskRuntime(700, 3500).longTask(Severity.SEVERE, 9, + Seq("Stage 6 has a long median task run time of 11.67 min.", + "Stage 6 has 3 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.")) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 6 has skew in task run time (median is 11.67 min, max is 58.33 min).", + "Stage 6: please try to modify the application to make the partitions more even.")).create(), + StageAnalysisBuilder(7, 2).taskRuntime(700, 900).longTask(Severity.SEVERE, 6, + Seq("Stage 7 has a long median task run time of 11.67 min.", + "Stage 7 has 2 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 7: please increase the number of partitions.")).create(), + StageAnalysisBuilder(8, 3).taskRuntime(3000, 3000).longTask(Severity.CRITICAL, 12, + Seq("Stage 8 has a long median task run time of 50.00 min.", + "Stage 8 has 3 tasks, 5 TB input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 8: please try to reduce the amount of data being processed.")) + .input(5 << 20).create(), + StageAnalysisBuilder(9, 4003).taskRuntime(3000, 3000).longTask(Severity.CRITICAL, 16012, + Seq("Stage 9 has a long median task run time of 50.00 min.", + "Stage 9 has 4003 tasks, 0 B input, 5 TB shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 9: please try to reduce the amount of data being processed.")) + .shuffleRead(5 << 20).create(), + StageAnalysisBuilder(10, 4000).taskRuntime(700, 900).longTask(Severity.SEVERE, 12000, + Seq("Stage 10 has a long median task run time of 11.67 min.", + "Stage 10 has 4000 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 10: please increase the number of partitions.")).create()) + + val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data) + val stageAnalysis = stageAnalyzer.getStageAnalysis() + (0 until expectedAnalysis.size).foreach { i => + compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i)) + } + } + + it("has execution memory spill severity") { + val heuristicConfigurationData = createHeuristicConfigurationData() + val stages = Seq( + StageBuilder(1, 5).taskRuntime(100, 150, 400).shuffleRead(200, 300, 800) + .spill(1, 2, 5).create(), + StageBuilder(2, 5).taskRuntime(100, 150, 400).shuffleRead(200, 300, 800) + .spill(10, 15, 40).create(), + StageBuilder(3, 5).taskRuntime(100, 150, 400).input(500, 2000, 3000) + .spill(100, 150, 400).create(), + StageBuilder(4, 5).taskRuntime(300, 350, 1500).shuffleWrite(1000, 1000, 5000) + .spill(300, 350, 1500).create(), + StageBuilder(5, 5).taskRuntime(300, 2500, 3000).shuffleRead(1000, 5000, 16000) + .shuffleWrite(300, 2500, 3000).spill(300, 2500, 3000).create(), + StageBuilder(6, 3).taskRuntime(50, 250, 350).input(50, 250, 350) + .spill(250, 250, 750).create(), + StageBuilder(7, 3).taskRuntime(50, 250, 350).output(250, 1000, 1500) + .spill(250, 250, 750).create(), + StageBuilder(8, 5).taskRuntime(2, 50, 53) + .times("09/09/2018 12:00:00", "09/09/2018 12:01:00") + .shuffleRead(500, 500, 1500).spill(250, 250, 750).create(), + StageBuilder(9, 5).taskRuntime(50, 250, 350).output(50, 250, 6L << 20) + .spill(50, 250, 2L << 20).create(), + StageBuilder(10, 5).taskRuntime(50, 250, 350).input(50, 250, 6L << 20) + .spill(50, 250, 2L << 20).create(), + StageBuilder(11, 3).taskRuntime(50, 250, 350).input(50, 250, 6L << 20) + .spill(50, 250, 3L << 20).create(), + StageBuilder(12, 3).taskRuntime(50, 250, 350).output(50, 250, 6L << 20) + .spill(50, 250, 4L << 20).create()) + val properties = Map( "spark.sql.shuffle.partitions" -> "5") + val data = createSparkApplicationData(stages, Seq.empty, Some(properties)) + + val expectedAnalysis = Seq( + StageAnalysisBuilder(1, 5).taskRuntime(100, 150).shuffleRead(800) + .spill(Severity.NONE, Severity.NONE, 0, 2, 5, Seq()).create(), + StageAnalysisBuilder(2, 5).taskRuntime(100, 150).shuffleRead(800) + .spill(Severity.LOW, Severity.LOW, 0, 15, 40, Seq()).create(), + StageAnalysisBuilder(3, 5).taskRuntime(100, 150).input(3000) + .spill(Severity.MODERATE, Severity.MODERATE, 10, 150, 400, + Seq("Stage 3 has 400 MB execution memory spill.")) + .skew(Severity.NONE, Severity.NONE, 0, + Seq("Stage 3 has skew in task input bytes (median is 500 MB, max is 1.95 GB).", + "Stage 3: please try to modify the application to make the input partitions more even.")) + .create(), + StageAnalysisBuilder(4, 5).taskRuntime(300, 350).shuffleWrite(5000) + .longTask(Severity.MODERATE, 10, + Seq("Stage 4 has a long median task run time of 5.00 min.", + "Stage 4 has 5 tasks, 0 B input, 0 B shuffle read, 4.88 GB shuffle write, and 0 B output.")) + .spill(Severity.SEVERE, Severity.SEVERE, 15, 350, 1500, + Seq("Stage 4 has 1.46 GB execution memory spill.")).create(), + StageAnalysisBuilder(5, 5).taskRuntime(300, 2500).shuffleRead(16000).shuffleWrite(3000) + .longTask(Severity.MODERATE, 10, Seq("Stage 5 has a long median task run time of 5.00 min.", + "Stage 5 has 5 tasks, 0 B input, 15.62 GB shuffle read, 2.93 GB shuffle write, and 0 B output.")) + .skew(Severity.SEVERE, Severity.SEVERE, 15, + Seq("Stage 5 has skew in task run time (median is 5.00 min, max is 41.67 min).", + "Stage 5 has skew in memory bytes spilled (median is 300 MB, max is 2.44 GB).", + "Stage 5 has skew in task shuffle read bytes (median is 1,000 MB, max is 4.88 GB).", + "Stage 5 has skew in task shuffle write bytes (median is 300 MB, max is 2.44 GB).", + "Stage 5: please try to modify the application to make the partitions more even.")) + .spill(Severity.MODERATE, Severity.MODERATE, 10, 2500, 3000 + , Seq("Stage 5 has 2.93 GB execution memory spill.")).create(), + StageAnalysisBuilder(6, 3).taskRuntime(50, 250).input(350) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 6 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 6 has skew in task input bytes (median is 50 MB, max is 250 MB).", + "Stage 6: please try to modify the application to make the input partitions more even.")) + .spill(Severity.CRITICAL, Severity.CRITICAL, 12, 250, 750, + Seq("Stage 6 has 750 MB execution memory spill.")).create(), + StageAnalysisBuilder(7, 3).taskRuntime(50, 250).output(1500) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 7 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 7 has skew in task output bytes (median is 250 MB, max is 1,000 MB).", + "Stage 7: please try to modify the application to make the partitions more even.")) + .spill(Severity.CRITICAL, Severity.CRITICAL, 12, 250, 750, + Seq("Stage 7 has 750 MB execution memory spill.")).create(), + StageAnalysisBuilder(8, 5).taskRuntime(2, 50).duration(60).shuffleRead(1500) + .skew(Severity.CRITICAL, Severity.NONE, 0, + Seq("Stage 8: please try to modify the application to make the partitions more even.")) + .spill(Severity.CRITICAL, Severity.CRITICAL, 20, 250, 750, + Seq("Stage 8 has 750 MB execution memory spill.")).create(), + StageAnalysisBuilder(9, 5).taskRuntime(50, 250).output(6L << 20) + .skew(Severity.MODERATE, Severity.MODERATE, 10, + Seq("Stage 9 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 9 has skew in memory bytes spilled (median is 50 MB, max is 250 MB).", + "Stage 9 has skew in task output bytes (median is 50 MB, max is 250 MB).", + "Stage 9: please try to modify the application to make the partitions more even.") + ) + .spill(Severity.SEVERE, Severity.NONE, 0, 250, 2L << 20, + Seq("Stage 9: a large amount of data is being processesd. Examine the application to see if this can be reduced.", + "Stage 9 has 2 TB execution memory spill.", + "Stage 9 has 5 tasks, 0 B input read, 0 B shuffle read, 0 B shuffle write, 6 TB output.", + "Stage 9 has median task values: 50 MB memory spill, 0 B input, 0 B shuffle read, 0 B shuffle write, 50 MB output.")) + .create(), + StageAnalysisBuilder(10, 5).taskRuntime(50, 250).input(6 << 20) + .skew(Severity.MODERATE, Severity.MODERATE, 10, + Seq("Stage 10 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 10 has skew in memory bytes spilled (median is 50 MB, max is 250 MB).", + "Stage 10 has skew in task input bytes (median is 50 MB, max is 250 MB).", + "Stage 10: please try to modify the application to make the input partitions more even.")) + .spill(Severity.SEVERE, Severity.NONE, 0, 250, 2L << 20, + Seq("Stage 10: a large amount of data is being processesd. Examine the application to see if this can be reduced.", + "Stage 10 has 2 TB execution memory spill.", + "Stage 10 has 5 tasks, 6 TB input read, 0 B shuffle read, 0 B shuffle write, 0 B output.", + "Stage 10 has median task values: 50 MB memory spill, 50 MB input, 0 B shuffle read, 0 B shuffle write, 0 B output.")) + .create(), + StageAnalysisBuilder(11, 3).taskRuntime(50, 250).input(6 << 20) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 11 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 11 has skew in memory bytes spilled (median is 50 MB, max is 250 MB).", + "Stage 11 has skew in task input bytes (median is 50 MB, max is 250 MB).", + "Stage 11: please try to modify the application to make the input partitions more even.")) + .spill(Severity.CRITICAL, Severity.NONE, 0, 250, 3L << 20, + Seq("Stage 11: a large amount of data is being processesd. Examine the application to see if this can be reduced.", + "Stage 11 has 3 TB execution memory spill.", + "Stage 11 has 3 tasks, 6 TB input read, 0 B shuffle read, 0 B shuffle write, 0 B output.", + "Stage 11 has median task values: 50 MB memory spill, 50 MB input, 0 B shuffle read, 0 B shuffle write, 0 B output.")) + .create(), + StageAnalysisBuilder(12, 3).taskRuntime(50, 250).output(6L << 20) + .skew(Severity.MODERATE, Severity.MODERATE, 6, + Seq("Stage 12 has skew in task run time (median is 50.00 sec, max is 4.17 min).", + "Stage 12 has skew in memory bytes spilled (median is 50 MB, max is 250 MB).", + "Stage 12 has skew in task output bytes (median is 50 MB, max is 250 MB).", + "Stage 12: please try to modify the application to make the partitions more even.")) + .spill(Severity.CRITICAL, Severity.NONE, 0, 250, 4L << 20, + Seq("Stage 12: a large amount of data is being processesd. Examine the application to see if this can be reduced.", + "Stage 12 has 4 TB execution memory spill.", + "Stage 12 has 3 tasks, 0 B input read, 0 B shuffle read, 0 B shuffle write, 6 TB output.", + "Stage 12 has median task values: 50 MB memory spill, 0 B input, 0 B shuffle read, 0 B shuffle write, 50 MB output.")) + .create()) + + val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data) + val stageAnalysis = stageAnalyzer.getStageAnalysis() + (0 until expectedAnalysis.size).foreach { i => + compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i)) + } + } + } + + it("custom recommendations") { + val heuristicConfigurationData = createHeuristicConfigurationData( + Map("execution_memory_spill_large_data_recommendation" -> "please try to filter the data to reduce the size", + "task_skew_input_data_recommendation" -> "please set DaliSpark.SPLIT_SIZE to make partitions more even", + "task_skew_generic_recommendation" -> "please make the partitions more even", + "long_tasks_large_data_recommendation" -> "please try to filter the data to reduce the size and increase speed", + "slow_tasks_recommendation" -> "optimize the code to increase speed", + "long tasks_few_partitions" -> "increase the number of partitions to speed up the stage", + "long tasks_few_input_partitions" -> "please set DaliSpark.SPLIT_SIZE to make partitions more even")) + val stages = Seq( + StageBuilder(1, 4003).taskRuntime(3000, 3000, 9000).shuffleRead(2 << 20, 3 << 20, 5 << 20).create(), + StageBuilder(2, 4000).taskRuntime(700, 900, 2000).create(), + StageBuilder(3, 2).taskRuntime(700, 900, 2000).create(), + StageBuilder(4, 3).taskRuntime(3000, 3000, 9000).input(2 << 20, 3 << 20, 5 << 20).create(), + StageBuilder(5, 3).taskRuntime(5, 250, 70).create(), + StageBuilder(6, 3).taskRuntime(20, 250, 300).input(20, 250, 300).create(), + StageBuilder(9, 5).taskRuntime(50, 50, 350).output(250, 250, 6L << 20) + .spill(250, 250, 2L << 20).create()) + val properties = Map( "spark.sql.shuffle.partitions" -> "3") + val data = createSparkApplicationData(stages, Seq.empty, Some(properties)) + + val expectedAnalysis = Seq( + StageAnalysisBuilder(1, 4003).taskRuntime(3000, 3000).longTask(Severity.CRITICAL, 16012, + Seq("Stage 1 has a long median task run time of 50.00 min.", + "Stage 1 has 4003 tasks, 0 B input, 5 TB shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 1: please try to filter the data to reduce the size and increase speed.")) + .shuffleRead(5 << 20).create(), + StageAnalysisBuilder(2, 4000).taskRuntime(700, 900).longTask(Severity.SEVERE, 12000, + Seq("Stage 2 has a long median task run time of 11.67 min.", + "Stage 2 has 4000 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 2: increase the number of partitions to speed up the stage.")).create(), + StageAnalysisBuilder(3, 2).taskRuntime(700, 900).longTask(Severity.SEVERE, 6, + Seq("Stage 3 has a long median task run time of 11.67 min.", + "Stage 3 has 2 tasks, 0 B input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 3: increase the number of partitions to speed up the stage.")).create(), + StageAnalysisBuilder(4, 3).taskRuntime(3000, 3000).longTask(Severity.CRITICAL, 12, + Seq("Stage 4 has a long median task run time of 50.00 min.", + "Stage 4 has 3 tasks, 5 TB input, 0 B shuffle read, 0 B shuffle write, and 0 B output.", + "Stage 4: please try to filter the data to reduce the size and increase speed.")) + .input(5 << 20).create(), + StageAnalysisBuilder(5, 3).taskRuntime(5, 250) + .skew(Severity.CRITICAL, Severity.CRITICAL, 12, + Seq("Stage 5 has skew in task run time (median is 5.00 sec, max is 4.17 min).", + "Stage 5: please make the partitions more even.") + ).create(), + StageAnalysisBuilder(6, 3).taskRuntime(20, 250).input(300) + .skew(Severity.SEVERE, Severity.SEVERE, 9, + Seq("Stage 6 has skew in task run time (median is 20.00 sec, max is 4.17 min).", + "Stage 6 has skew in task input bytes (median is 20 MB, max is 250 MB).", + "Stage 6: please set DaliSpark.SPLIT_SIZE to make partitions more even.")).create(), + StageAnalysisBuilder(7, 5).taskRuntime(50, 50).output(6L << 20) + .spill(Severity.SEVERE, Severity.NONE, 0, 250, 2L << 20, + Seq("Stage 9: please try to filter the data to reduce the size.", + "Stage 9 has 2 TB execution memory spill.", + "Stage 9 has 5 tasks, 0 B input read, 0 B shuffle read, 0 B shuffle write, 6 TB output.", + "Stage 9 has median task values: 250 MB memory spill, 0 B input, 0 B shuffle read, 0 B shuffle write, 250 MB output.")) + .create()) + + val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data) + val stageAnalysis = stageAnalyzer.getStageAnalysis() + (0 until expectedAnalysis.size).foreach { i => + compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i)) + } + } + + /** compare actual and expected StageAnalysis */ + private def compareStageAnalysis(actual: StageAnalysis, expected: StageAnalysis): Unit = { + compareExecutionMemorySpillResult(actual.executionMemorySpillResult, expected.executionMemorySpillResult) + compareSimpleStageAnalysisResult(actual.longTaskResult, expected.longTaskResult) + compareTaskSkewResult(actual.taskSkewResult, expected.taskSkewResult) + compareTaskFailureResult(actual.taskFailureResult, expected.taskFailureResult) + compareSimpleStageAnalysisResult(actual.stageFailureResult, expected.stageFailureResult) + actual.numTasks should be (expected.numTasks) + actual.medianRunTime should be (expected.medianRunTime) + actual.maxRunTime should be (expected.maxRunTime) + actual.stageDuration should be (expected.stageDuration) + actual.inputBytes should be(expected.inputBytes) + actual.outputBytes should be(expected.outputBytes) + actual.shuffleReadBytes should be(expected.shuffleReadBytes) + actual.shuffleWriteBytes should be(expected.shuffleWriteBytes) + } + + /** compare actual and expected ExecutionMemorySpillResult */ + private def compareExecutionMemorySpillResult( + actual: ExecutionMemorySpillResult, + expected: ExecutionMemorySpillResult) = { + actual.severity should be(expected.severity) + actual.rawSeverity should be(expected.rawSeverity) + actual.score should be(expected.score) + actual.memoryBytesSpilled should be(expected.memoryBytesSpilled) + actual.maxTaskBytesSpilled should be(expected.maxTaskBytesSpilled) + actual.details should be(expected.details) + } + + /** compare actual and expected SimpleStageAnalysisResult */ + private def compareSimpleStageAnalysisResult( + actual: SimpleStageAnalysisResult, + expected: SimpleStageAnalysisResult) = { + actual.severity should be(expected.severity) + actual.score should be(expected.score) + actual.details should be(expected.details) + } + + /** compare actual and expected TaskSkewResult */ + private def compareTaskSkewResult( + actual: TaskSkewResult, + expected: TaskSkewResult) = { + actual.severity should be(expected.severity) + actual.rawSeverity should be(expected.rawSeverity) + actual.score should be(expected.score) + actual.details should be(expected.details) + } + + /** compare actual and expected TaskFailureResult */ + private def compareTaskFailureResult( + actual: TaskFailureResult, + expected: TaskFailureResult) = { + actual.severity should be(expected.severity) + actual.oomSeverity should be(expected.oomSeverity) + actual.containerKilledSeverity should be(expected.containerKilledSeverity) + actual.score should be(expected.score) + actual.numFailures should be(expected.numFailures) + actual.numOOM should be(expected.numOOM) + actual.numContainerKilled should be (expected.numContainerKilled) + actual.details should be(expected.details) + } +} diff --git a/test/com/linkedin/drelephant/spark/heuristics/StagesHeuristicTest.scala b/test/com/linkedin/drelephant/spark/heuristics/StagesHeuristicTest.scala index e6aae4fe1..cd1acc3d0 100644 --- a/test/com/linkedin/drelephant/spark/heuristics/StagesHeuristicTest.scala +++ b/test/com/linkedin/drelephant/spark/heuristics/StagesHeuristicTest.scala @@ -133,10 +133,16 @@ object StagesHeuristicTest { status, stageId, attemptId = 0, + numTasks = numCompleteTasks + numFailedTasks, numActiveTasks = numCompleteTasks + numFailedTasks, numCompleteTasks, numFailedTasks, executorRunTime, + executorCpuTime = 0, + submissionTime = None, + firstTaskLaunchedTime = None, + completionTime = None, + failureReason = None, inputBytes = 0, inputRecords = 0, outputBytes = 0, @@ -152,7 +158,13 @@ object StagesHeuristicTest { schedulingPool = "", accumulatorUpdates = Seq.empty, tasks = None, - executorSummary = None + executorSummary = None, + peakJvmUsedMemory = None, + peakExecutionMemory = None, + peakStorageMemory = None, + peakUnifiedMemory = None, + taskSummary = None, + executorMetricsSummary = None ) def newFakeSparkApplicationData( diff --git a/test/com/linkedin/drelephant/spark/heuristics/StagesWithFailedTasksHeuristicTest.scala b/test/com/linkedin/drelephant/spark/heuristics/StagesWithFailedTasksHeuristicTest.scala index cdfdc11ea..7a6191395 100644 --- a/test/com/linkedin/drelephant/spark/heuristics/StagesWithFailedTasksHeuristicTest.scala +++ b/test/com/linkedin/drelephant/spark/heuristics/StagesWithFailedTasksHeuristicTest.scala @@ -97,11 +97,18 @@ object StagesWithFailedTasksHeuristicTest { status, stageId, attemptId = 0, + numTasks = 0, numActiveTasks = numCompleteTasks, numCompleteTasks, numFailedTasks = 3, executorRunTime = 0, - inputBytes = 0, + executorCpuTime = 0, + submissionTime = None, + firstTaskLaunchedTime = None, + completionTime = None, + failureReason = None, + + inputBytes = 0, inputRecords = 0, outputBytes = 0, outputRecords = 0, @@ -150,7 +157,13 @@ object StagesWithFailedTasksHeuristicTest { errorMessage = None, taskMetrics = None) )), - executorSummary = None + executorSummary = None, + peakJvmUsedMemory = None, + peakExecutionMemory = None, + peakStorageMemory = None, + peakUnifiedMemory = None, + taskSummary = None, + executorMetricsSummary = None ) def newFakeSparkApplicationData diff --git a/test/com/linkedin/drelephant/util/UtilsTest.java b/test/com/linkedin/drelephant/util/UtilsTest.java index 69f5509c4..8cef6c2ea 100644 --- a/test/com/linkedin/drelephant/util/UtilsTest.java +++ b/test/com/linkedin/drelephant/util/UtilsTest.java @@ -197,6 +197,17 @@ public void testGetDurationBreakdown() { assertEquals("0:05:24", Utils.getDurationBreakdown(durations[4])); assertEquals("314483:43:34", Utils.getDurationBreakdown(durations[5])); } + + @Test + public void testGetDuration() { + long []durations = { 153, 25431, 432344, 23423562, 178123456L}; + assertEquals("153 ms", Utils.getDuration(durations[0])); + assertEquals("25.43 sec", Utils.getDuration(durations[1])); + assertEquals("7.21 min", Utils.getDuration(durations[2])); + assertEquals("6.51 hr", Utils.getDuration(durations[3])); + assertEquals("2.06 days", Utils.getDuration(durations[4])); + } + @Test public void testGetPercentage() { long []numerators = {10,20,30,40,50};