iRakson commented on a change in pull request #26756:
[SPARK-30119][WebUI]Support Pagination for Completed Batch Table in Streaming
Tab
URL: https://github.com/apache/spark/pull/26756#discussion_r354124299
##########
File path:
streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala
##########
@@ -156,40 +160,234 @@ private[ui] class ActiveBatchTable(
}
}
-private[ui] class CompletedBatchTable(batches: Seq[BatchUIData],
batchInterval: Long)
- extends BatchTableBase("completed-batches-table", batchInterval) {
+private[ui] class CompletedBatchTableRow(val batchData: BatchUIData,
+ val batchTime: Long,
+ val numRecords: Long,
+ val schedulingDelay: Option[Long],
+ val processingDelay: Option[Long],
+ val totalDelay: Option[Long])
- private val firstFailureReason = getFirstFailureReason(batches)
+private[ui] class CompletedBatchPagedTable(
+ request: HttpServletRequest,
+ parent: StreamingTab,
+ batchInterval: Long,
+ data: Seq[BatchUIData],
+ completedBatchTag: String,
+ basePath: String,
+ subPath: String,
+ parameterOtherTable: Iterable[String],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean)
+ extends PagedTable[CompletedBatchTableRow] {
- override protected def columns: Seq[Node] = super.columns ++ {
- <th>Total Delay {SparkUIUtils.tooltip("Total time taken to handle a
batch", "top")}</th>
- <th>Output Ops: Succeeded/Total</th> ++ {
- if (firstFailureReason.nonEmpty) {
- <th>Error</th>
- } else {
- Nil
- }
- }
+ override val dataSource = new CompletedBatchTableDataSource(data, pageSize,
sortColumn, desc)
+
+ private val parameterPath =
s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}"
+
+ private val firstFailureReason = getFirstFailureReason(data)
+
+ override def tableId: String = completedBatchTag
+
+ 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"&$completedBatchTag.sort=$encodedSortColumn" +
+ s"&$completedBatchTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize"
}
- override protected def renderRows: Seq[Node] = {
- batches.flatMap(batch => <tr>{completedBatchRow(batch)}</tr>)
+ override def pageSizeFormField: String = s"$completedBatchTag.pageSize"
+
+ override def pageNumberFormField: String = s"$completedBatchTag.page"
+
+ override def goButtonFormPath: String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
s"$parameterPath&$completedBatchTag.sort=$encodedSortColumn&$completedBatchTag.desc=$desc"
}
- private def completedBatchRow(batch: BatchUIData): Seq[Node] = {
- val totalDelay = batch.totalDelay
+ override def headers: Seq[Node] = {
+ val completedBatchTableHeaders = Seq("Batch Time", "Records", "Scheduling
Delay",
+ "Processing Delay", "Total Delay", "Output Ops: Succeeded/Total")
+
+ val tooltips = Seq(None, None, Some("Time taken by Streaming scheduler to"
+
+ " submit jobs of a batch"), Some("Time taken to process all jobs of a
batch"),
+ Some("Total time taken to handle a batch"), None)
+
+ assert(completedBatchTableHeaders.length == tooltips.length)
Review comment:
assert is there to make sure of that only.
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
