[SPARK-3454] Expose JSON representation of data shown in WebU

core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala

@@ -19,13 +19,33 @@ package org.apache.spark.ui.env
 
 import javax.servlet.http.HttpServletRequest
 
+import org.json4s.{JObject, JValue}
+import org.json4s.JsonDSL._
+
 import scala.xml.Node
 
 import org.apache.spark.ui.{UIUtils, WebUIPage}
 
 private[ui] class EnvironmentPage(parent: EnvironmentTab) extends WebUIPage("") {
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    val jvmInfoJson =
+      ("RUntime Informationf" -> listener.jvmInformation.foldLeft(JObject())(_ ~ _))
+    val sparkPropertiesJson =
+      ("Spark Properties" -> listener.sparkProperties.foldLeft(JObject())(_ ~ _))
+    val systemPropertiesJson =
+      ("System Properties" -> listener.systemProperties.foldLeft(JObject())(_ ~ _))
+    val classPathEntriesJson =
+      ("Classpath Entries" -> listener.classpathEntries.foldLeft(JObject())(_ ~ _))
+
+    val environment = ("Environment" -> jvmInfoJson ~
+                                        sparkPropertiesJson ~
+                                        systemPropertiesJson ~
+                                        classPathEntriesJson)
+    environment
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     val runtimeInformationTable = UIUtils.listingTable(
       propertyHeader, jvmRow, listener.jvmInformation, fixedWidth = true)

core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala

@@ -21,6 +21,9 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
+import org.json4s.JValue
+import org.json4s.JsonDSL._
+
 import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
 import org.apache.spark.util.Utils
 
@@ -44,6 +47,30 @@ private case class ExecutorSummaryInfo(
 private[ui] class ExecutorsPage(parent: ExecutorsTab) extends WebUIPage("") {
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    val storageStatusList = listener.storageStatusList
+
+    val execInfoJsonList = for (statusId <- 0 until storageStatusList.size) yield  {
+      val execInfo = getExecInfo(statusId)
+      ("Executor ID" -> execInfo.id) ~
+      ("Address" -> execInfo.hostPort) ~
+      ("RDD Blocks" -> execInfo.rddBlocks) ~
+      ("Memory Used" -> execInfo.memoryUsed) ~
+      ("Disk Used" -> execInfo.diskUsed) ~
+      ("Active Tasks" -> execInfo.activeTasks) ~
+      ("Failed Tasks" -> execInfo.failedTasks) ~
+      ("Complete Tasks" -> execInfo.completedTasks) ~
+      ("TotalTasks" -> execInfo.totalTasks) ~
+      ("Task Time" -> execInfo.totalDuration) ~
+      ("Input" -> execInfo.totalInputBytes) ~
+      ("Shuffle Read" -> execInfo.totalShuffleRead) ~
+      ("Shuffle Write" -> execInfo.totalShuffleWrite)
+    }
+
+    ("Executor List" -> execInfoJsonList)
+
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     val storageStatusList = listener.storageStatusList
     val maxMem = storageStatusList.map(_.maxMem).sum

core/src/main/scala/org/apache/spark/ui/jobs/JobProgressPage.scala

@@ -19,9 +19,14 @@ package org.apache.spark.ui.jobs
 
 import javax.servlet.http.HttpServletRequest
 
+import org.apache.spark.util.JsonProtocol
+import org.json4s.JValue
+import org.json4s.JsonAST.JNothing
+import org.json4s.JsonDSL._
+
 import scala.xml.{Node, NodeSeq}
 
-import org.apache.spark.scheduler.Schedulable
+import org.apache.spark.scheduler.{StageInfo, Schedulable}
 import org.apache.spark.ui.{WebUIPage, UIUtils}
 
 /** Page showing list of all ongoing and recently finished stages and pools */
@@ -31,6 +36,35 @@ private[ui] class JobProgressPage(parent: JobProgressTab) extends WebUIPage("")
   private val listener = parent.listener
   private lazy val isFairScheduler = parent.isFairScheduler
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    listener.synchronized {
+
+      val activeStageList = listener.activeStages.values.map {
+        case info: StageInfo =>
+          JsonProtocol.stageInfoToJson(info)
+      }
+      val activeStageJson = ("Active Stages" -> activeStageList)
+
+      val completedStageList = listener.completedStages.reverse.map {
+        case info: StageInfo =>
+          JsonProtocol.stageInfoToJson(info)
+      }
+      val completedStageJson = ("Completed Stages" -> completedStageList)
+
+      val failedStageList = listener.failedStages.reverse.map {
+        case info: StageInfo =>
+          JsonProtocol.stageInfoToJson(info)
+      }
+      val failedStageJson = ("Failed Stages" -> failedStageList)
+
+      ("Stages Info" ->
+        ("Scheduling Mode" -> listener.schedulingMode.map(_.toString).getOrElse("Unknown")) ~
+        activeStageJson ~
+        completedStageJson ~
+        failedStageJson)
+    }
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     listener.synchronized {
       val activeStages = listener.activeStages.values.toSeq

core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala

@@ -19,8 +19,13 @@ package org.apache.spark.ui.jobs
 
 import javax.servlet.http.HttpServletRequest
 
+import org.apache.spark.util.JsonProtocol
+
 import scala.xml.Node
 
+import org.json4s.JValue
+import org.json4s.JsonDSL._
+
 import org.apache.spark.scheduler.{Schedulable, StageInfo}
 import org.apache.spark.ui.{WebUIPage, UIUtils}
 
@@ -30,6 +35,35 @@ private[ui] class PoolPage(parent: JobProgressTab) extends WebUIPage("pool") {
   private val sc = parent.sc
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    listener.synchronized {
+      val poolName = request.getParameter("poolname")
+      val poolToActiveStages = listener.poolToActiveStages
+      val activeStages = poolToActiveStages.get(poolName) match {
+        case Some(s) => s.values.map {
+          case info: StageInfo =>
+            JsonProtocol.stageInfoToJson(info)
+        }
+        case None => Seq[JValue]()
+      }
+
+      val pools = if (live) Seq(sc.getPoolForName(poolName).get) else Seq[Schedulable]()
+
+      val poolList = pools.map {
+        case schedulable: Schedulable =>
+          ("Pool Name" -> schedulable.name) ~
+          ("Minimum Share" -> schedulable.minShare) ~
+          ("Pool Weight" -> schedulable.weight) ~
+          ("Active Stages" -> activeStages) ~
+          ("Running Tasks" -> schedulable.runningTasks) ~
+          ("Scheduling Mode" -> schedulable.schedulingMode.toString)
+      }
+
+      ("Pools" -> poolList)
+
+    }
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     listener.synchronized {
       val poolName = request.getParameter("poolname")

core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala

@@ -20,17 +20,76 @@ package org.apache.spark.ui.jobs
 import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
+import org.json4s.JsonAST.JNothing
+
 import scala.xml.{Node, Unparsed}
 
+import org.json4s.{JObject, JValue}
+import org.json4s.JsonDSL._
+
 import org.apache.spark.ui.{ToolTips, WebUIPage, UIUtils}
 import org.apache.spark.ui.jobs.UIData._
-import org.apache.spark.util.{Utils, Distribution}
+import org.apache.spark.util.{JsonProtocol, Utils, Distribution}
 import org.apache.spark.scheduler.AccumulableInfo
 
 /** Page showing statistics and task list for a given stage */
 private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    val stageId = request.getParameter("id").toInt
+    val stageAttemptId = request.getParameter("attempt").toInt
+
+    var stageSummary = ("Stage ID" -> stageId) ~ ("Stage Attempt ID" -> stageAttemptId)
+
+
+    val stageDataOpt = listener.stageIdToData.get((stageId, stageAttemptId))
+    var retVal: JValue = JNothing
+
+    if (!stageDataOpt.isEmpty && !stageDataOpt.get.taskData.isEmpty) {
+      val stageData = stageDataOpt.get
+
+      stageSummary ~= ("Executor Run Time" -> stageData.executorRunTime)
+      if (stageData.inputBytes > 0) stageSummary ~= ("Input Bytes" -> stageData.inputBytes)
+      if (stageData.shuffleReadBytes > 0) {
+        stageSummary ~= ("Shuffle Read Bytes" -> stageData.shuffleReadBytes)
+      }
+
+      if (stageData.shuffleWriteBytes > 0) {
+        stageSummary ~= ("Shuffle Write bytes" -> stageData.shuffleWriteBytes)
+      }
+
+      if (stageData.memoryBytesSpilled > 0 && stageData.diskBytesSpilled > 0) {
+        stageSummary ~=
+          ("Memory Bytes Spilled" -> stageData.memoryBytesSpilled) ~
+          ("Disk Bytes Spilled" -> stageData.diskBytesSpilled)
+      }
+
+      val tasks = stageData.taskData.values.toSeq.sortBy(_.taskInfo.launchTime)
+
+      val taskList = tasks.map {
+        case uiData: TaskUIData =>
+          var jsonTaskInfo: JValue = JsonProtocol.taskInfoToJson(uiData.taskInfo)
+          val jsonTaskMetrics: JValue =
+            if (uiData.taskMetrics.isDefined) {
+              JsonProtocol.taskMetricsToJson(uiData.taskMetrics.get)
+            } else JNothing
+
+          if (jsonTaskInfo.isInstanceOf[JObject] && jsonTaskMetrics.isInstanceOf[JObject]) {
+            jsonTaskInfo =
+              jsonTaskInfo.asInstanceOf[JObject] ~ jsonTaskMetrics.asInstanceOf[JObject]
+          }
+          jsonTaskInfo
+      }
+
+      retVal =
+        ("Stage Info" ->
+          ("StageSummary" -> stageSummary) ~
+          ("Tasks" -> taskList))
+    }
+    retVal
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     listener.synchronized {
       val stageId = request.getParameter("id").toInt

core/src/main/scala/org/apache/spark/ui/storage/RDDPage.scala

@@ -19,16 +19,77 @@ package org.apache.spark.ui.storage
 
 import javax.servlet.http.HttpServletRequest
 
+import org.json4s.JsonAST.JNothing
+
 import scala.xml.Node
 
-import org.apache.spark.storage.{BlockId, BlockStatus, StorageStatus, StorageUtils}
+import org.json4s.JValue
+import org.json4s.JsonDSL._
+
+import org.apache.spark.storage._
 import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
 /** Page showing storage details for a given RDD */
 private[ui] class RDDPage(parent: StorageTab) extends WebUIPage("rdd") {
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    val rddId = request.getParameter("id").toInt
+    val storageStatusList = listener.storageStatusList
+    val rddInfoOpt = listener.rddInfoList.find(_.id == rddId)
+
+    var retVal: JValue = JNothing
+
+    if (rddInfoOpt.isDefined) {
+      val rddInfo = rddInfoOpt.get
+
+      val rddSummaryJson = ("RDD Summary" ->
+        ("RDD ID" -> rddId) ~
+        ("Storage Level" -> rddInfo.storageLevel.description) ~
+        ("Cached Partitions" -> rddInfo.numCachedPartitions) ~
+        ("Total Partitions" -> rddInfo.numPartitions) ~
+        ("Memory Size" -> rddInfo.memSize) ~
+        ("Disk Size" -> rddInfo.diskSize))
+
+      val dataDistributionList = storageStatusList.map {
+        case status: StorageStatus =>
+          ("Host" -> (status.blockManagerId.host + ":" + status.blockManagerId.port)) ~
+          ("Memory Usage" -> status.memUsedByRdd(rddId)) ~
+          ("Memory Remaining" -> status.memRemaining) ~
+          ("Disk Usage" -> status.diskUsedByRdd(rddId))
+      }
+
+      val dataDistributionJson = ("Data Distribution" -> dataDistributionList)
+
+      val blockLocations = StorageUtils.getRddBlockLocations(rddId, storageStatusList)
+      val blocks = storageStatusList
+        .flatMap(_.rddBlocksById(rddId))
+        .sortWith(_._1.name < _._1.name)
+        .map { case (blockId, status) =>
+          (blockId, status, blockLocations.get(blockId).getOrElse(Seq[String]("Unknown")))
+      }
+      val partitionList = blocks.map {
+        case (id: BlockId, block: BlockStatus, locations: Seq[String]) =>
+          ("Block Name" -> id.toString) ~
+          ("Storage Level" -> block.storageLevel.description) ~
+          ("Size in Memory" -> block.memSize) ~
+          ("Size on Disk" -> block.diskSize) ~
+          ("Executors" -> locations)
+      }
+      val partitionsJson = ("Partitions" -> partitionList)
+
+
+      retVal =
+        ("RDD Info" ->
+          rddSummaryJson ~
+          dataDistributionJson ~
+          partitionsJson
+        )
+    }
+    retVal
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     val rddId = request.getParameter("id").toInt
     val storageStatusList = listener.storageStatusList

core/src/main/scala/org/apache/spark/ui/storage/StoragePage.scala

@@ -21,14 +21,26 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
+import org.json4s.JValue
+import org.json4s.JsonDSL._
+
 import org.apache.spark.storage.RDDInfo
 import org.apache.spark.ui.{WebUIPage, UIUtils}
-import org.apache.spark.util.Utils
+import org.apache.spark.util.{JsonProtocol, Utils}
 
 /** Page showing list of RDD's currently stored in the cluster */
 private[ui] class StoragePage(parent: StorageTab) extends WebUIPage("") {
   private val listener = parent.listener
 
+  override def renderJson(request: HttpServletRequest): JValue = {
+    val rddList = listener.rddInfoList.map {
+      case info: RDDInfo =>
+        JsonProtocol.rddInfoToJson(info)
+    }
+
+    ("RDDs" -> rddList)
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     val rdds = listener.rddInfoList
     val content = UIUtils.listingTable(rddHeader, rddRow, rdds)