Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Initial changes for adding Spark stage analysis #438

Open
wants to merge 8 commits into
base: customSHSWork
Choose a base branch
from
Open

Initial changes for adding Spark stage analysis #438

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
dca51b8
Initial changes for adding Spark stage analysis.
edwinalu Sep 21, 2018
aafd7d4
Add keys for thresholds, comments, getDuration test.
edwinalu Sep 24, 2018
2f6adbf
factor out results for different heuristics
edwinalu Sep 25, 2018
fbd4e79
code review comments
edwinalu Sep 25, 2018
af09da0
lower DEFAULT_MAX_DATA_PROCESSED_THRESHOLD
edwinalu Sep 27, 2018
c5d421a
move shared values to StageAnalysisResult
edwinalu Oct 8, 2018
7966c61
check fetchFailedTasks parameter
edwinalu Oct 15, 2018
135bb05
code review comments
edwinalu Oct 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
code review comments
  • Loading branch information
@edwinalu
edwinalu committed Oct 24, 2018
commit fbd4e791bd26f14bd9f625de71a5882c098ed8b2
33 changes: 14 additions & 19 deletions app/com/linkedin/drelephant/spark/heuristics/StagesAnalyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,17 +81,11 @@ private[heuristics] class StagesAnalyzer(
private val maxRecommendedPartitions = heuristicConfigurationData.getParamMap
.getOrDefault(MAX_RECOMMENDED_PARTITIONS_KEY, DEFAULT_MAX_RECOMMENDED_PARTITIONS).toInt

/**
* Get the analysis for each stage of the application.
*
* @param curNumPartitions the configured number of partitions for the application
* (value of spark.sql.shuffle.partitions).
* @return list of analysis results of stages.
*/
def getStageAnalysis(curNumPartitions: Int): Seq[StageAnalysis] = {
val failedTasksStageMap = data.stagesWithFailedTasks.flatMap { stageData =>
stageData.tasks.map(tasks => (stageData.stageId, tasks.values))
}.toMap
/** @return list of analysis results for all the stages. */
def getStageAnalysis(): Seq[StageAnalysis] = {
val appConfigurationProperties: Map[String, String] = data.appConfigurationProperties
val curNumPartitions = appConfigurationProperties.get(SPARK_SQL_SHUFFLE_PARTITIONS)
.map(_.toInt).getOrElse(SPARK_SQL_SHUFFLE_PARTITIONS_DEFAULT)

data.stageDatas.map { stageData =>
val medianTime = stageData.taskSummary.collect {
Expand All @@ -112,7 +106,7 @@ private[heuristics] class StagesAnalyzer(
val taskSkewResult = checkForTaskSkew(stageId, stageData, medianTime, maxTime, stageDuration,
executionMemorySpillResult.severity)
val stageFailureResult = checkForStageFailure(stageId, stageData)
val taskFailureResult = checkForTaskFailure(stageId, stageData, failedTasksStageMap)
val taskFailureResult = checkForTaskFailure(stageId, stageData)
val gcResult = checkForGC(stageId, stageData)

new StageAnalysis(stageData.stageId, executionMemorySpillResult, longTaskResult,
Expand Down Expand Up @@ -210,7 +204,6 @@ private[heuristics] class StagesAnalyzer(
taskSkewThresholds.severityOf(max / median)
case _ => Severity.NONE
}
val median = medianTime.getOrElse(0.0D)
val maximum = maxTime.getOrElse(0.0D)
val taskSkewSeverity =
if (maximum > taskDurationMinThreshold &&
Expand All @@ -225,12 +218,12 @@ private[heuristics] class StagesAnalyzer(
// add more information about what might be causing skew if skew is being flagged
// (reported severity is significant), or there is execution memory spill, since skew
// can also cause execution memory spill.
val median = Utils.getDuration(medianTime.map(_.toLong).getOrElse(0L))
val maximum = Utils.getDuration(maxTime.map(_.toLong).getOrElse(0L))
val medianStr = Utils.getDuration(medianTime.map(_.toLong).getOrElse(0L))
val maximumStr = Utils.getDuration(maxTime.map(_.toLong).getOrElse(0L))
var inputSkewSeverity = Severity.NONE
if (hasSignificantSeverity(taskSkewSeverity)) {
details +=
s"Stage $stageId has skew in task run time (median is $median, max is $maximum)"
s"Stage $stageId has skew in task run time (median is $medianStr, max is $maximumStr)"
}
stageData.taskSummary.foreach { summary =>
checkSkewedData(stageId, summary.memoryBytesSpilled(DISTRIBUTION_MEDIAN_IDX),
Expand Down Expand Up @@ -367,13 +360,15 @@ private[heuristics] class StagesAnalyzer(
*
* @param stageId stage ID.
* @param stageData stage data.
* @param failedTasksStageMap map of stage ID to list of failed tasks for the stage.
* @return result of failed tasks analysis for the stage.
*/
private def checkForTaskFailure(
stageId: Int,
stageData: StageData,
failedTasksStageMap: Map[Int, Iterable[TaskDataImpl]]): TaskFailureResult = {
stageData: StageData): TaskFailureResult = {
val failedTasksStageMap = data.stagesWithFailedTasks.flatMap { stageData =>
stageData.tasks.map(tasks => (stageData.stageId, tasks.values))
}.toMap

val failedTasks = failedTasksStageMap.get(stageId)

val details = new ArrayBuffer[String]()
Expand Down
20 changes: 12 additions & 8 deletions test/com/linkedin/drelephant/spark/heuristics/StagesAnalyzerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
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 data = createSparkApplicationData(stages, Seq.empty, None)
val properties = Map( "spark.sql.shuffle.partitions" -> "200")
val data = createSparkApplicationData(stages, Seq.empty, Some(properties))

val expectedAnalysis = Seq(
StageAnalysisBuilder(1, 3).create(),
Expand All @@ -48,7 +49,7 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
.create())

val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data)
val stageAnalysis = stageAnalyzer.getStageAnalysis(200)
val stageAnalysis = stageAnalyzer.getStageAnalysis()
(0 until expectedAnalysis.size).foreach { i =>
compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i))
}
Expand All @@ -71,7 +72,8 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
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 data = createSparkApplicationData(stages, Seq.empty, None)
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)
Expand Down Expand Up @@ -130,7 +132,7 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
Seq()).create())

val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data)
val stageAnalysis = stageAnalyzer.getStageAnalysis(5)
val stageAnalysis = stageAnalyzer.getStageAnalysis()
(0 until expectedAnalysis.size).foreach { i =>
compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i))
}
Expand All @@ -147,8 +149,9 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
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())
val properties = Map( "spark.sql.shuffle.partitions" -> "3")
val data = createSparkApplicationData(stages, Seq.empty, Some(properties))

val data = createSparkApplicationData(stages, Seq.empty, None)
val expectedAnalysis = Seq(
StageAnalysisBuilder(1, 3).taskRuntime(120, 150).create(),
StageAnalysisBuilder(2, 3).taskRuntime(180, 200).longTask(Severity.LOW, 0, Seq()).create(),
Expand All @@ -172,7 +175,7 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
.input(5 << 20).create())

val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data)
val stageAnalysis = stageAnalyzer.getStageAnalysis(3)
val stageAnalysis = stageAnalyzer.getStageAnalysis()
(0 until expectedAnalysis.size).foreach { i =>
compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i))
}
Expand Down Expand Up @@ -206,7 +209,8 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
.spill(50, 250, 3L << 20).create(),
StageBuilder(12, 3).taskRuntime(50, 250, 350).output(50, 250, 6 << 20)
.spill(50, 250, 4L << 20).create())
val data = createSparkApplicationData(stages, Seq.empty, None)
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)
Expand Down Expand Up @@ -304,7 +308,7 @@ class StagesAnalyzerTest extends FunSpec with Matchers {
.create())

val stageAnalyzer = new StagesAnalyzer(heuristicConfigurationData, data)
val stageAnalysis = stageAnalyzer.getStageAnalysis(5)
val stageAnalysis = stageAnalyzer.getStageAnalysis()
(0 until expectedAnalysis.size).foreach { i =>
compareStageAnalysis(stageAnalysis(i), expectedAnalysis(i))
}
Expand Down