Skip to content

SPARK-2533 - Add locality levels on stage summary view #9487

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

Closed
wants to merge 7 commits into from
Closed

SPARK-2533 - Add locality levels on stage summary view #9487

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f02d53c
SPARK-2533 - Add locality levels on stage summary view
jbonofre Nov 5, 2015
e3b1478
SPARK-2533 - Update locality level format and move in the "Show Addit…
jbonofre Nov 5, 2015
9a1e27b
SPARK-2533 - Update locality level summary string format according to…
jbonofre Nov 5, 2015
439671d
SPARK-2533 - Remove unused import
jbonofre Nov 5, 2015
4d41ce9
SPARK-2533 - Alphabetize import
jbonofre Nov 5, 2015
e9abe3f
SPARK-2533 - Update locality level summary string format according to…
jbonofre Nov 6, 2015
bd7de49
SPARK-2533 - Add locality levels summary on the stage view
jbonofre Nov 12, 2015
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
21 changes: 20 additions & 1 deletion core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.apache.commons.lang3.StringEscapeUtils

import org.apache.spark.{InternalAccumulator, SparkConf}
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.scheduler.{AccumulableInfo, TaskInfo}
import org.apache.spark.scheduler.{AccumulableInfo, TaskInfo, TaskLocality}
import org.apache.spark.ui._
import org.apache.spark.ui.jobs.UIData._
import org.apache.spark.util.{Utils, Distribution}
Expand Down Expand Up @@ -70,6 +70,21 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {

private val displayPeakExecutionMemory = parent.conf.getBoolean("spark.sql.unsafe.enabled", true)

private def getLocalitySummaryString(stageData: StageUIData): String = {
val localities = stageData.taskData.values.map(_.taskInfo.taskLocality)
val localityCounts = localities.groupBy(identity).mapValues(_.size)
val localityNamesAndCounts = localityCounts.toSeq.map { case (locality, count) =>
val localityName = locality match {
case TaskLocality.PROCESS_LOCAL => "Process local"
case TaskLocality.NODE_LOCAL => "Node local"
case TaskLocality.RACK_LOCAL => "Rack local"
case TaskLocality.ANY => "Any"
}
s"$localityName: $count"
}
localityNamesAndCounts.sorted.mkString("; ")
}

def render(request: HttpServletRequest): Seq[Node] = {
progressListener.synchronized {
val parameterId = request.getParameter("id")
Expand Down Expand Up @@ -129,6 +144,10 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
<strong>Total Time Across All Tasks: </strong>
{UIUtils.formatDuration(stageData.executorRunTime)}
</li>
<li>
<strong>Locality Level Summary: </strong>
{getLocalitySummaryString(stageData)}
</li>
{if (stageData.hasInput) {
<li>
<strong>Input Size / Records: </strong>
Expand Down