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_r354668482
##########
File path:
streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala
##########
@@ -156,40 +160,232 @@ 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)
- 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
- }
- }
+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 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)
+
+ val headerRow: Seq[Node] = {
+ completedBatchTableHeaders.zip(tooltips).map { case (header, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$completedBatchTag.sort=${URLEncoder.encode(header,
UTF_8.name())}" +
+ s"&$completedBatchTag.desc=${!desc}" +
+ s"&$completedBatchTag.pageSize=$pageSize" +
+ s"#$completedBatchTag")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ if (tooltip.nonEmpty) {
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" title={tooltip.get}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <a href={headerLink}>
+ {header} {Unparsed(arrow)}
+ </a>
+ </th>
+ }
+ } else {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$completedBatchTag.sort=${URLEncoder.encode(header,
UTF_8.name())}" +
+ s"&$completedBatchTag.pageSize=$pageSize" +
+ s"#$completedBatchTag")
+
+ if(tooltip.nonEmpty) {
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" title={tooltip.get}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <a href={headerLink}>
+ {header}
+ </a>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ {headerRow}
+ </thead>
+ }
+
+ override def row(completedBatchRow: CompletedBatchTableRow): Seq[Node] = {
+ val batch = completedBatchRow.batchData
+ val batchTime = completedBatchRow.batchTime
+ val formattedBatchTime = UIUtils.formatBatchTime(batchTime, batchInterval)
+ val numRecords = completedBatchRow.numRecords
+ val schedulingDelay = completedBatchRow.schedulingDelay
+ val formattedSchedulingDelay =
schedulingDelay.map(SparkUIUtils.formatDuration).getOrElse("-")
+ val processingTime = completedBatchRow.processingDelay
+ val formattedProcessingTime =
processingTime.map(SparkUIUtils.formatDuration).getOrElse("-")
+ val batchTimeId = s"batch-$batchTime"
+ val totalDelay = completedBatchRow.totalDelay
val formattedTotalDelay =
totalDelay.map(SparkUIUtils.formatDuration).getOrElse("-")
- baseRow(batch) ++ {
+ <tr>
+ <td id={batchTimeId} sorttable_customkey={batchTime.toString}>
+ <a href={s"batch?id=$batchTime"}>
+ {formattedBatchTime}
+ </a>
+ </td>
+ <td sorttable_customkey={numRecords.toString}>{numRecords.toString}
records</td>
Review comment:
No. Not required. Removed this.
----------------------------------------------------------------
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
