From 7be05201239c9fec7133e8f09455224786e91b59 Mon Sep 17 00:00:00 2001 From: pgandhi Date: Wed, 18 Jul 2018 13:31:20 -0500 Subject: [PATCH 1/9] [SPARK-24851] : Map a Stage ID to it's Associated Job ID in UI Added a field in Stage UI to display the corresponding job id for that particular stage. --- .../scala/org/apache/spark/status/AppStatusStore.scala | 7 +++++++ .../main/scala/org/apache/spark/ui/jobs/StagePage.scala | 8 ++++++++ 2 files changed, 15 insertions(+) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index e237281c552b1..db99d8d85ac10 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -94,6 +94,13 @@ private[spark] class AppStatusStore( }.toSeq } + def getJobIdsAssociatedWithStage(stageId: Int): Seq[Set[Int]] = { + store.view(classOf[StageDataWrapper]).index("stageId").first(stageId).last(stageId) + .asScala.map { s => + s.jobIds + }.toSeq + } + def lastStageAttempt(stageId: Int): v1.StageData = { val it = store.view(classOf[StageDataWrapper]) .index("stageId") diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index 55eb989962668..472bd38c6e763 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -115,6 +115,8 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We return UIUtils.headerSparkPage(request, stageHeader, content, parent) } + val stageJobIds = parent.store.getJobIdsAssociatedWithStage(stageId) + val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) val totalTasks = taskCount(stageData) @@ -182,6 +184,12 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We {Utils.bytesToString(stageData.diskBytesSpilled)} }} + {if (!stageJobIds.isEmpty) { +
  • + Associated Job Ids: + {stageJobIds} +
    • + }} From 3151a62091abec423867b889cabe137dd939ef12 Mon Sep 17 00:00:00 2001 From: pgandhi Date: Mon, 23 Jul 2018 13:42:14 -0500 Subject: [PATCH 2/9] [SPARK-24851] : Using both stage id as well as the attempt id to fetch the job ids for the particular stage --- .../scala/org/apache/spark/status/AppStatusStore.scala | 9 ++++----- .../main/scala/org/apache/spark/ui/jobs/StagePage.scala | 2 +- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index db99d8d85ac10..d444c54a91316 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -94,11 +94,10 @@ private[spark] class AppStatusStore( }.toSeq } - def getJobIdsAssociatedWithStage(stageId: Int): Seq[Set[Int]] = { - store.view(classOf[StageDataWrapper]).index("stageId").first(stageId).last(stageId) - .asScala.map { s => - s.jobIds - }.toSeq + def getJobIdsAssociatedWithStage(stageId: Int, stageAttemptId: Int): Seq[Int] = { + val stageKey = Array(stageId, stageAttemptId) + val jobIds = store.read(classOf[StageDataWrapper], stageKey).jobIds.toSeq + jobIds } def lastStageAttempt(stageId: Int): v1.StageData = { diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index 472bd38c6e763..d4dfa4ef86168 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -115,7 +115,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We return UIUtils.headerSparkPage(request, stageHeader, content, parent) } - val stageJobIds = parent.store.getJobIdsAssociatedWithStage(stageId) + val stageJobIds = parent.store.getJobIdsAssociatedWithStage(stageId, stageAttemptId) val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) From a50e8b17fcca5a194434aca73ecdf2422925c15e Mon Sep 17 00:00:00 2001 From: pgandhi Date: Wed, 25 Jul 2018 13:34:33 -0500 Subject: [PATCH 3/9] [SPARK-24851] : Avoiding extra store lookup and adding href links --- .../org/apache/spark/status/AppStatusStore.scala | 15 ++++++--------- .../spark/status/api/v1/StagesResource.scala | 2 +- .../org/apache/spark/ui/jobs/StagePage.scala | 12 ++++++++---- 3 files changed, 15 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index d444c54a91316..cc6c541bc5a7f 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -94,12 +94,6 @@ private[spark] class AppStatusStore( }.toSeq } - def getJobIdsAssociatedWithStage(stageId: Int, stageAttemptId: Int): Seq[Int] = { - val stageKey = Array(stageId, stageAttemptId) - val jobIds = store.read(classOf[StageDataWrapper], stageKey).jobIds.toSeq - jobIds - } - def lastStageAttempt(stageId: Int): v1.StageData = { val it = store.view(classOf[StageDataWrapper]) .index("stageId") @@ -118,10 +112,13 @@ private[spark] class AppStatusStore( } } - def stageAttempt(stageId: Int, stageAttemptId: Int, details: Boolean = false): v1.StageData = { + def stageAttempt(stageId: Int, stageAttemptId: Int, details: Boolean = false): StageDataWrapper = { val stageKey = Array(stageId, stageAttemptId) - val stage = store.read(classOf[StageDataWrapper], stageKey).info - if (details) stageWithDetails(stage) else stage + val stageDataWrapper: StageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) + val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info + val jobIds = stageDataWrapper.jobIds + val returnStageDataWrapper: StageDataWrapper = new StageDataWrapper(stage, jobIds, null) + returnStageDataWrapper } def taskCount(stageId: Int, stageAttemptId: Int): Long = { diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala index 96249e4bfd5fa..c7775523431df 100644 --- a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala +++ b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala @@ -56,7 +56,7 @@ private[v1] class StagesResource extends BaseAppResource { @PathParam("stageAttemptId") stageAttemptId: Int, @QueryParam("details") @DefaultValue("true") details: Boolean): StageData = withUI { ui => try { - ui.store.stageAttempt(stageId, stageAttemptId, details = details) + ui.store.stageAttempt(stageId, stageAttemptId, details = details).info } catch { case _: NoSuchElementException => // Change the message depending on whether there are any attempts for the requested stage. diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index d4dfa4ef86168..f4309ce9bece6 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -105,8 +105,9 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We val stageAttemptId = parameterAttempt.toInt val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)" + val stageDataWrapper = parent.store.stageAttempt(stageId, stageAttemptId, details = false) val stageData = parent.store - .asOption(parent.store.stageAttempt(stageId, stageAttemptId, details = false)) + .asOption(stageDataWrapper.info) .getOrElse { val content =
    @@ -115,7 +116,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We return UIUtils.headerSparkPage(request, stageHeader, content, parent) } - val stageJobIds = parent.store.getJobIdsAssociatedWithStage(stageId, stageAttemptId) + val stageJobIds = stageDataWrapper.jobIds.toSeq val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) @@ -187,7 +188,10 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We {if (!stageJobIds.isEmpty) {
  • Associated Job Ids: - {stageJobIds} + {for(jobId <- stageJobIds) yield {val detailUrl = "%s/jobs/job/?id=%s".format( + UIUtils.prependBaseUri(request, parent.basePath), jobId) + {s"${jobId}"}    + }}
    • }} @@ -1055,7 +1059,7 @@ private[ui] object ApiHelper { } def lastStageNameAndDescription(store: AppStatusStore, job: JobData): (String, String) = { - val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0)) + val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0).info) (stage.map(_.name).getOrElse(""), stage.flatMap(_.description).getOrElse(job.name)) } From d57e6dcdf0e5c54e09cbd9b47b3ad4a38eb393cb Mon Sep 17 00:00:00 2001 From: pgandhi Date: Wed, 25 Jul 2018 13:44:47 -0500 Subject: [PATCH 4/9] [SPARK-24851] : Fixing Scalastyle Tests --- .../main/scala/org/apache/spark/status/AppStatusStore.scala | 3 ++- core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index cc6c541bc5a7f..7ca79c48325f9 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -112,7 +112,8 @@ private[spark] class AppStatusStore( } } - def stageAttempt(stageId: Int, stageAttemptId: Int, details: Boolean = false): StageDataWrapper = { + def stageAttempt(stageId: Int, stageAttemptId: Int, + details: Boolean = false): StageDataWrapper = { val stageKey = Array(stageId, stageAttemptId) val stageDataWrapper: StageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index f4309ce9bece6..0251deda238bd 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -188,7 +188,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We {if (!stageJobIds.isEmpty) {
  • Associated Job Ids: - {for(jobId <- stageJobIds) yield {val detailUrl = "%s/jobs/job/?id=%s".format( + {for (jobId <- stageJobIds) yield {val detailUrl = "%s/jobs/job/?id=%s".format( UIUtils.prependBaseUri(request, parent.basePath), jobId) {s"${jobId}"}    }} From 3a06b876268c291604c395eef51387a419486fee Mon Sep 17 00:00:00 2001 From: pgandhi Date: Thu, 26 Jul 2018 16:33:58 -0500 Subject: [PATCH 5/9] [SPARK-24851] : Adding exception handling part to avoid NoSuchElementException on querying an invalid stage id --- .../org/apache/spark/ui/jobs/StagePage.scala | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index 0251deda238bd..6fc3737daa911 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -18,7 +18,7 @@ package org.apache.spark.ui.jobs import java.net.URLEncoder -import java.util.Date +import java.util.{Date, NoSuchElementException} import java.util.concurrent.TimeUnit import javax.servlet.http.HttpServletRequest @@ -105,16 +105,29 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We val stageAttemptId = parameterAttempt.toInt val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)" - val stageDataWrapper = parent.store.stageAttempt(stageId, stageAttemptId, details = false) - val stageData = parent.store - .asOption(stageDataWrapper.info) - .getOrElse { + var stageDataWrapper: StageDataWrapper = null + try { + stageDataWrapper = parent.store.stageAttempt(stageId, stageAttemptId, details = false) + } catch { + case e: NoSuchElementException => e.getMessage + } + var stageData: StageData = null + if (stageDataWrapper != null) { + stageData = parent.store + .asOption(stageDataWrapper.info) + .get + } else { + stageData = { val content =
    -

    No information to display for Stage {stageId} (Attempt {stageAttemptId})

    +

    No information to display for Stage + {stageId} + (Attempt + {stageAttemptId})

    return UIUtils.headerSparkPage(request, stageHeader, content, parent) } + } val stageJobIds = stageDataWrapper.jobIds.toSeq From 52b08b2d42645257cea2e51bcbb4fdbdf4794936 Mon Sep 17 00:00:00 2001 From: pgandhi Date: Fri, 27 Jul 2018 14:32:11 -0500 Subject: [PATCH 6/9] [SPARK-24851] : Addressing reviews July 27 Using map instead of for..yield and returning tuple instead of object --- .../apache/spark/status/AppStatusStore.scala | 5 ++--- .../spark/status/api/v1/StagesResource.scala | 2 +- .../org/apache/spark/ui/jobs/StagePage.scala | 22 +++++++++---------- 3 files changed, 13 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index 7ca79c48325f9..80aadf9d5ed7e 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -113,13 +113,12 @@ private[spark] class AppStatusStore( } def stageAttempt(stageId: Int, stageAttemptId: Int, - details: Boolean = false): StageDataWrapper = { + details: Boolean = false): (v1.StageData, Seq[Int]) = { val stageKey = Array(stageId, stageAttemptId) val stageDataWrapper: StageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info val jobIds = stageDataWrapper.jobIds - val returnStageDataWrapper: StageDataWrapper = new StageDataWrapper(stage, jobIds, null) - returnStageDataWrapper + (stage, jobIds.toSeq) } def taskCount(stageId: Int, stageAttemptId: Int): Long = { diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala index c7775523431df..30d52b97833e6 100644 --- a/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala +++ b/core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala @@ -56,7 +56,7 @@ private[v1] class StagesResource extends BaseAppResource { @PathParam("stageAttemptId") stageAttemptId: Int, @QueryParam("details") @DefaultValue("true") details: Boolean): StageData = withUI { ui => try { - ui.store.stageAttempt(stageId, stageAttemptId, details = details).info + ui.store.stageAttempt(stageId, stageAttemptId, details = details)._1 } catch { case _: NoSuchElementException => // Change the message depending on whether there are any attempts for the requested stage. diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index 6fc3737daa911..7d09207bf19e6 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -105,18 +105,15 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We val stageAttemptId = parameterAttempt.toInt val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)" - var stageDataWrapper: StageDataWrapper = null + var stageDataTuple: Tuple2[StageData, Seq[Int]] = null try { - stageDataWrapper = parent.store.stageAttempt(stageId, stageAttemptId, details = false) + stageDataTuple = parent.store.stageAttempt(stageId, stageAttemptId, details = false) } catch { case e: NoSuchElementException => e.getMessage } var stageData: StageData = null - if (stageDataWrapper != null) { - stageData = parent.store - .asOption(stageDataWrapper.info) - .get - } else { + var stageJobIds: Seq[Int] = null + if (stageDataTuple == null) { stageData = { val content =
    @@ -127,10 +124,11 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
    return UIUtils.headerSparkPage(request, stageHeader, content, parent) } + } else { + stageData = stageDataTuple._1 + stageJobIds = stageDataTuple._2 } - val stageJobIds = stageDataWrapper.jobIds.toSeq - val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) val totalTasks = taskCount(stageData) @@ -201,10 +199,10 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We {if (!stageJobIds.isEmpty) {
  • Associated Job Ids: - {for (jobId <- stageJobIds) yield {val detailUrl = "%s/jobs/job/?id=%s".format( + {stageJobIds.map(jobId => {val detailUrl = "%s/jobs/job/?id=%s".format( UIUtils.prependBaseUri(request, parent.basePath), jobId) {s"${jobId}"}    - }} + })}
    • }} @@ -1072,7 +1070,7 @@ private[ui] object ApiHelper { } def lastStageNameAndDescription(store: AppStatusStore, job: JobData): (String, String) = { - val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0).info) + val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0)._1) (stage.map(_.name).getOrElse(""), stage.flatMap(_.description).getOrElse(job.name)) } From 9eff537eb268853d4cc1e6e16d4f27d598b46490 Mon Sep 17 00:00:00 2001 From: pgandhi Date: Thu, 9 Aug 2018 09:53:05 -0500 Subject: [PATCH 7/9] [SPARK-24851] : Addressing Reviews August 9 --- .../apache/spark/status/AppStatusStore.scala | 4 +-- .../org/apache/spark/ui/jobs/StagePage.scala | 35 ++++++++++--------- 2 files changed, 20 insertions(+), 19 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index 80aadf9d5ed7e..8468eda0802a2 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -113,9 +113,9 @@ private[spark] class AppStatusStore( } def stageAttempt(stageId: Int, stageAttemptId: Int, - details: Boolean = false): (v1.StageData, Seq[Int]) = { + details: Boolean = false): (v1.StageData, Seq[Int]) = { val stageKey = Array(stageId, stageAttemptId) - val stageDataWrapper: StageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) + val stageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info val jobIds = stageDataWrapper.jobIds (stage, jobIds.toSeq) diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index 7d09207bf19e6..e9e16ab690700 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -105,28 +105,29 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We val stageAttemptId = parameterAttempt.toInt val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)" - var stageDataTuple: Tuple2[StageData, Seq[Int]] = null - try { - stageDataTuple = parent.store.stageAttempt(stageId, stageAttemptId, details = false) + var stageDataTuple: Option[Tuple2[StageData, Seq[Int]]] = try { + Some(parent.store.stageAttempt(stageId, stageAttemptId, details = false)) } catch { case e: NoSuchElementException => e.getMessage + None } var stageData: StageData = null var stageJobIds: Seq[Int] = null - if (stageDataTuple == null) { - stageData = { - val content = -
    -

    No information to display for Stage - {stageId} - (Attempt - {stageAttemptId})

    -
    - return UIUtils.headerSparkPage(request, stageHeader, content, parent) - } - } else { - stageData = stageDataTuple._1 - stageJobIds = stageDataTuple._2 + stageDataTuple match { + case Some(stageTuple) => + stageData = stageTuple._1 + stageJobIds = stageTuple._2 + case None => + stageData = { + val content = +
    +

    No information to display for Stage + {stageId} + (Attempt + {stageAttemptId})

    +
    + return UIUtils.headerSparkPage(request, stageHeader, content, parent) + } } val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) From 3b7384011c0d4e41ecd5c4b797f08c1c62088fbd Mon Sep 17 00:00:00 2001 From: pgandhi Date: Fri, 28 Sep 2018 10:15:37 -0500 Subject: [PATCH 8/9] [SPARK-24851] : Addressing reviews September 28, 2018 Simplifying code block in StagePage --- .../org/apache/spark/ui/jobs/StagePage.scala | 35 ++++++------------- 1 file changed, 10 insertions(+), 25 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala index e9e16ab690700..c5e68e49cfb95 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala @@ -18,7 +18,7 @@ package org.apache.spark.ui.jobs import java.net.URLEncoder -import java.util.{Date, NoSuchElementException} +import java.util.Date import java.util.concurrent.TimeUnit import javax.servlet.http.HttpServletRequest @@ -105,30 +105,15 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We val stageAttemptId = parameterAttempt.toInt val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)" - var stageDataTuple: Option[Tuple2[StageData, Seq[Int]]] = try { - Some(parent.store.stageAttempt(stageId, stageAttemptId, details = false)) - } catch { - case e: NoSuchElementException => e.getMessage - None - } - var stageData: StageData = null - var stageJobIds: Seq[Int] = null - stageDataTuple match { - case Some(stageTuple) => - stageData = stageTuple._1 - stageJobIds = stageTuple._2 - case None => - stageData = { - val content = -
    -

    No information to display for Stage - {stageId} - (Attempt - {stageAttemptId})

    -
    - return UIUtils.headerSparkPage(request, stageHeader, content, parent) - } - } + val (stageData, stageJobIds) = parent.store + .asOption(parent.store.stageAttempt(stageId, stageAttemptId, details = false)) + .getOrElse { + val content = +
    +

    No information to display for Stage {stageId} (Attempt {stageAttemptId})

    +
    + return UIUtils.headerSparkPage(request, stageHeader, content, parent) + } val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId) From 0be099a81bb016034e73562cc32599bd06810956 Mon Sep 17 00:00:00 2001 From: pgandhi Date: Fri, 5 Oct 2018 15:01:05 -0500 Subject: [PATCH 9/9] [SPARK-24851] : Addressing Reviews October 5, 2018 --- .../main/scala/org/apache/spark/status/AppStatusStore.scala | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala index 8468eda0802a2..9839cbb99f862 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala @@ -117,8 +117,7 @@ private[spark] class AppStatusStore( val stageKey = Array(stageId, stageAttemptId) val stageDataWrapper = store.read(classOf[StageDataWrapper], stageKey) val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info - val jobIds = stageDataWrapper.jobIds - (stage, jobIds.toSeq) + (stage, stageDataWrapper.jobIds.toSeq) } def taskCount(stageId: Int, stageAttemptId: Int): Long = {