shahidki31 commented on a change in pull request #26756:
[SPARK-30119][WebUI]Support Pagination for Batch Tables in Streaming Tab
URL: https://github.com/apache/spark/pull/26756#discussion_r358111312
##########
File path:
streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala
##########
@@ -17,39 +17,122 @@
package org.apache.spark.streaming.ui
-import scala.xml.Node
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets.UTF_8
+import javax.servlet.http.HttpServletRequest
+
+import scala.xml.{Node, Unparsed}
+
+import org.apache.spark.ui.{PagedDataSource, PagedTable, UIUtils =>
SparkUIUtils}
+
+private[ui] class StreamingBatchPagedTable(
+ request: HttpServletRequest,
+ parent: StreamingTab,
+ batchInterval: Long,
+ batchData: Seq[BatchUIData],
+ streamingBatchTag: String,
+ basePath: String,
+ subPath: String,
+ parameterOtherTable: Iterable[String],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean) extends PagedTable[BatchUIData] {
+
+ override val dataSource = new StreamingBatchTableDataSource(batchData,
pageSize, sortColumn, desc)
+ private val parameterPath =
s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}"
+ private val firstFailureReason = getFirstFailureReason(batchData)
+
+ override def tableId: String = streamingBatchTag
+
+ override def tableCssClass: String =
+ "table table-bordered table-condensed table-striped " +
+ "table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$streamingBatchTag.sort=$encodedSortColumn" +
+ s"&$streamingBatchTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize"
+ }
-import org.apache.spark.ui.{UIUtils => SparkUIUtils}
+ override def pageSizeFormField: String = s"$streamingBatchTag.pageSize"
-private[ui] abstract class BatchTableBase(tableId: String, batchInterval:
Long) {
+ override def pageNumberFormField: String = s"$streamingBatchTag.page"
- protected def columns: Seq[Node] = {
- <th>Batch Time</th>
- <th>Records</th>
- <th>Scheduling Delay
- {SparkUIUtils.tooltip("Time taken by Streaming scheduler to submit
jobs of a batch", "top")}
- </th>
- <th>Processing Time
- {SparkUIUtils.tooltip("Time taken to process all jobs of a batch",
"top")}</th>
+ override def goButtonFormPath: String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
s"$parameterPath&$streamingBatchTag.sort=$encodedSortColumn&$streamingBatchTag.desc=$desc"
}
- /**
- * Return the first failure reason if finding in the batches.
- */
- protected def getFirstFailureReason(batches: Seq[BatchUIData]):
Option[String] = {
- batches.flatMap(_.outputOperations.flatMap(_._2.failureReason)).headOption
- }
-
- protected def getFirstFailureTableCell(batch: BatchUIData): Seq[Node] = {
- val firstFailureReason =
batch.outputOperations.flatMap(_._2.failureReason).headOption
- firstFailureReason.map { failureReason =>
- val failureReasonForUI =
UIUtils.createOutputOperationFailureForUI(failureReason)
- UIUtils.failureReasonCell(
- failureReasonForUI, rowspan = 1, includeFirstLineInExpandDetails =
false)
- }.getOrElse(<td>-</td>)
+ override def headers: Seq[Node] = {
+ val completedBatchTableHeaders = Seq("Batch Time", "Records", "Scheduling
Delay",
Review comment:
From the attached screenshot, I can see `Total Delay` isn't there in the
`ActiveBatches` table and `Status` field isn't there in the `CompletedBatches`
table?
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
