Convert streaming ui to use new table builder.

JoshRosen · JoshRosen · commit 290f58b47fd1 · 2014-10-21T00:38:21.000-07:00
diff --git a/core/src/main/scala/org/apache/spark/ui/UITables.scala b/core/src/main/scala/org/apache/spark/ui/UITables.scala
@@ -229,7 +229,19 @@ private[spark] class UITableBuilder[T](fixedWidth: Boolean = false) {
    * Display a column of durations, in milliseconds, as human-readable strings, such as "12 s".
    */
   def durationCol(name: String)(fieldExtractor: T => Long): UITableBuilder[T] = {
-    col[Long](name, formatter = UIUtils.formatDuration)(fieldExtractor)
+    col[Long](name, formatter = UIUtils.formatDuration, sortKey = Some(_.toString))(fieldExtractor)
+  }
+
+  /**
+   * Display a column of optional durations, in milliseconds, as human-readable strings,
+   * such as "12 s".  If the duration is None, then '-' will be displayed.
+   */
+  def optDurationCol(name: String)(fieldExtractor: T => Option[Long]): UITableBuilder[T] = {
+    col[Option[Long]](
+      name,
+      formatter = { _.map(UIUtils.formatDuration).getOrElse("-")},
+      sortKey = Some(_.getOrElse("-").toString)
+    )(fieldExtractor)
   }
 
   def build: UITable[T] = {
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
@@ -25,15 +25,16 @@ import scala.xml.Node
 import org.apache.spark.Logging
 import org.apache.spark.ui._
 import org.apache.spark.ui.UIUtils._
-import org.apache.spark.util.Distribution
+import org.apache.spark.streaming.scheduler.ReceiverInfo
+
 
 /** Page for Spark Web UI that shows statistics of a streaming job */
 private[ui] class StreamingPage(parent: StreamingTab)
   extends WebUIPage("") with Logging {
 
   private val listener = parent.listener
   private val startTime = Calendar.getInstance().getTime()
-  private val emptyCell = "-"
+  private val empty = "-"
 
   /** Render the page */
   def render(request: HttpServletRequest): Seq[Node] = {
@@ -45,6 +46,19 @@ private[ui] class StreamingPage(parent: StreamingTab)
     UIUtils.headerSparkPage("Streaming", content, parent, Some(5000))
   }
 
+  private val batchStatsTable: UITable[(String, Option[Long], Option[Seq[Double]])] = {
+    val builder = new UITableBuilder[(String, Option[Long], Option[Seq[Double]])]()
+    import builder._
+    col("Metric") { _._1 }
+    optDurationCol("Last batch") { _._2 }
+    optDurationCol("Minimum") { _._3.map(_(0).toLong) }
+    optDurationCol("25th percentile") { _._3.map(_(1).toLong) }
+    optDurationCol("Median") { _._3.map(_(2).toLong) }
+    optDurationCol("75th percentile") { _._3.map(_(3).toLong) }
+    optDurationCol("Maximum") { _._3.map(_(4).toLong) }
+    build
+  }
+
   /** Generate basic stats of the streaming program */
   private def generateBasicStats(): Seq[Node] = {
     val timeSinceStart = System.currentTimeMillis() - startTime.getTime
@@ -75,37 +89,45 @@ private[ui] class StreamingPage(parent: StreamingTab)
     val receivedRecordDistributions = listener.receivedRecordsDistributions
     val lastBatchReceivedRecord = listener.lastReceivedBatchRecords
     val table = if (receivedRecordDistributions.size > 0) {
-      val headerRow = Seq(
-        "Receiver",
-        "Status",
-        "Location",
-        "Records in last batch\n[" + formatDate(Calendar.getInstance().getTime()) + "]",
-        "Minimum rate\n[records/sec]",
-        "Median rate\n[records/sec]",
-        "Maximum rate\n[records/sec]",
-        "Last Error"
-      )
-      val dataRows = (0 until listener.numReceivers).map { receiverId =>
-        val receiverInfo = listener.receiverInfo(receiverId)
-        val receiverName = receiverInfo.map(_.name).getOrElse(s"Receiver-$receiverId")
-        val receiverActive = receiverInfo.map { info =>
-          if (info.active) "ACTIVE" else "INACTIVE"
-        }.getOrElse(emptyCell)
-        val receiverLocation = receiverInfo.map(_.location).getOrElse(emptyCell)
-        val receiverLastBatchRecords = formatNumber(lastBatchReceivedRecord(receiverId))
-        val receivedRecordStats = receivedRecordDistributions(receiverId).map { d =>
-          d.getQuantiles(Seq(0.0, 0.5, 1.0)).map(r => formatNumber(r.toLong))
-        }.getOrElse {
-          Seq(emptyCell, emptyCell, emptyCell, emptyCell, emptyCell)
+      val tableRenderer: UITable[(Int, Option[ReceiverInfo])] = {
+        val builder = new UITableBuilder[(Int, Option[ReceiverInfo])]()
+        import builder._
+        col("Receiver") { case (receiverId, receiverInfo) =>
+          receiverInfo.map(_.name).getOrElse(s"Receiver-$receiverId")
+        }
+        col("Status") { case (_, receiverInfo) =>
+          receiverInfo.map { info => if (info.active) "ACTIVE" else "INACTIVE" }.getOrElse(empty)
+        }
+        col("Location") { case (_, receiverInfo) => receiverInfo.map(_.location).getOrElse(empty) }
+        col("Records in last batch\n[" + formatDate(Calendar.getInstance().getTime()) + "]") {
+          case (receiverId, _) => formatNumber(lastBatchReceivedRecord(receiverId))
         }
-        val receiverLastError = listener.receiverInfo(receiverId).map { info =>
-          val msg = s"${info.lastErrorMessage} - ${info.lastError}"
-          if (msg.size > 100) msg.take(97) + "..." else msg
-        }.getOrElse(emptyCell)
-        Seq(receiverName, receiverActive, receiverLocation, receiverLastBatchRecords) ++
-          receivedRecordStats ++ Seq(receiverLastError)
+        col("Minimum rate\n[records/sec]") {
+          case (receiverId, _) => receivedRecordDistributions(receiverId).map {
+            _.getQuantiles(Seq(0.0)).map(formatNumber).head
+          }.getOrElse(empty)
+        }
+        col("Median rate\n[records/sec]") {
+          case (receiverId, _) => receivedRecordDistributions(receiverId).map {
+            _.getQuantiles(Seq(0.5)).map(formatNumber).head
+          }.getOrElse(empty)
+        }
+        col("Maximum rate\n[records/sec]") {
+          case (receiverId, _) => receivedRecordDistributions(receiverId).map {
+            _.getQuantiles(Seq(1.0)).map(formatNumber).head
+          }.getOrElse(empty)
+        }
+        col("Last Error") {
+          case (_, receiverInfo) => receiverInfo.map { info =>
+            val msg = s"${info.lastErrorMessage} - ${info.lastError}"
+            if (msg.size > 100) msg.take(97) + "..." else msg
+          }.getOrElse(empty)
+        }
+        build
       }
-      Some(listingTable(headerRow, dataRows))
+
+      val dataRows = (0 until listener.numReceivers).map { id => (id, listener.receiverInfo(id)) }
+      Some(tableRenderer.render(dataRows))
     } else {
       None
     }
@@ -121,33 +143,20 @@ private[ui] class StreamingPage(parent: StreamingTab)
   private def generateBatchStatsTable(): Seq[Node] = {
     val numBatches = listener.retainedCompletedBatches.size
     val lastCompletedBatch = listener.lastCompletedBatch
+
     val table = if (numBatches > 0) {
-      val processingDelayQuantilesRow = {
-        Seq(
-          "Processing Time",
-          formatDurationOption(lastCompletedBatch.flatMap(_.processingDelay))
-        ) ++ getQuantiles(listener.processingDelayDistribution)
-      }
-      val schedulingDelayQuantilesRow = {
-        Seq(
-          "Scheduling Delay",
-          formatDurationOption(lastCompletedBatch.flatMap(_.schedulingDelay))
-        ) ++ getQuantiles(listener.schedulingDelayDistribution)
-      }
-      val totalDelayQuantilesRow = {
-        Seq(
-          "Total Delay",
-          formatDurationOption(lastCompletedBatch.flatMap(_.totalDelay))
-        ) ++ getQuantiles(listener.totalDelayDistribution)
-      }
-      val headerRow = Seq("Metric", "Last batch", "Minimum", "25th percentile",
-        "Median", "75th percentile", "Maximum")
-      val dataRows: Seq[Seq[String]] = Seq(
-        processingDelayQuantilesRow,
-        schedulingDelayQuantilesRow,
-        totalDelayQuantilesRow
+      val rows: Seq[(String, Option[Long], Option[Seq[Double]])] = Seq(
+        ("Processing Time",
+          lastCompletedBatch.flatMap(_.processingDelay),
+          listener.processingDelayDistribution.map(_.getQuantiles())),
+        ("Scheduling Delay",
+          lastCompletedBatch.flatMap(_.schedulingDelay),
+          listener.schedulingDelayDistribution.map(_.getQuantiles())),
+        ("Total Delay",
+          lastCompletedBatch.flatMap(_.totalDelay),
+          listener.totalDelayDistribution.map(_.getQuantiles()))
       )
-      Some(listingTable(headerRow, dataRows))
+      Some(batchStatsTable.render(rows))
     } else {
       None
     }
@@ -162,26 +171,4 @@ private[ui] class StreamingPage(parent: StreamingTab)
 
     content
   }
-
-
-  /**
-   * Returns a human-readable string representing a duration such as "5 second 35 ms"
-   */
-  private def formatDurationOption(msOption: Option[Long]): String = {
-    msOption.map(formatDurationVerbose).getOrElse(emptyCell)
-  }
-
-  /** Get quantiles for any time distribution */
-  private def getQuantiles(timeDistributionOption: Option[Distribution]) = {
-    timeDistributionOption.get.getQuantiles().map { ms => formatDurationVerbose(ms.toLong) }
-  }
-
-  /** Generate HTML table from string data */
-  private def listingTable(headers: Seq[String], data: Seq[Seq[String]]) = {
-    def generateDataRow(data: Seq[String]): Seq[Node] = {
-      <tr> {data.map(d => <td>{d}</td>)} </tr>
-    }
-    UIUtils.listingTable(headers, generateDataRow, data, fixedWidth = true)
-  }
 }
-
