[GitHub] [spark] ron8hu commented on a change in pull request #31204: [SPARK-26399][WEBUI][CORE] Add new stage-level REST APIs and parameters

[GitBox <gi...@apache.org>]

ron8hu commented on a change in pull request #31204:
URL: https://github.com/apache/spark/pull/31204#discussion_r570721165



##########
File path: core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
##########
@@ -138,14 +211,63 @@ private[spark] class AppStatusStore(
     }
   }
 
-  def stageAttempt(stageId: Int, stageAttemptId: Int,
-      details: Boolean = false): (v1.StageData, Seq[Int]) = {
+  def stageAttempt(
+      stageId: Int,
+      stageAttemptId: Int,
+      details: Boolean = false,
+      taskStatus: JList[v1.TaskStatus]): (v1.StageData, Seq[Int]) = {
     val stageKey = Array(stageId, stageAttemptId)
     val stageDataWrapper = store.read(classOf[StageDataWrapper], stageKey)
-    val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info
+    val stage = if (details) {
+      stageWithDetails(stageDataWrapper.info, taskStatus)
+    } else {
+      stageDataWrapper.info
+    }
     (stage, stageDataWrapper.jobIds.toSeq)
   }
 
+  def stageExecutorSummary(
+    stageId: Int,
+    stageAttemptId: Int,
+    unsortedQuantiles: Array[Double]): Option[v1.ExecutorMetricsDistributions] = {
+    val quantiles = unsortedQuantiles.sorted
+    val summary = executorSummary(stageId, stageAttemptId)
+    if (summary.isEmpty) {
+      None
+    } else {
+      val values = summary.values.toIndexedSeq
+
+      def getQuantilesValue(
+        values: IndexedSeq[Double],
+        quantiles: Array[Double]): IndexedSeq[Double] = {
+        val count = values.size
+        val indices = quantiles.map { q => math.min((q * count).toLong, count - 1) }
+        indices.map(i => values(i.toInt)).toIndexedSeq
+      }
+
+      Some(new v1.ExecutorMetricsDistributions(
+        quantiles = quantiles,
+        taskTime = getQuantilesValue(values.map(_.taskTime.toDouble).sorted, quantiles),
+        failedTasks = getQuantilesValue(values.map(_.taskTime.toDouble).sorted, quantiles),

Review comment:
       My co-worker @xkrogen spotted this error.  When you copied and pasted line 250, you forgot to change it to the right val field name.  It should be (_.failedTasks.toDouble).  Same for the following 12 lines.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org