shahidki31 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_r354841199
##########
File path:
streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala
##########
@@ -156,40 +160,209 @@ private[ui] class ActiveBatchTable(
}
}
-private[ui] class CompletedBatchTable(batches: Seq[BatchUIData],
batchInterval: Long)
- extends BatchTableBase("completed-batches-table", batchInterval) {
+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[BatchUIData] {
- private val firstFailureReason = getFirstFailureReason(batches)
+ override val dataSource = new CompletedBatchTableDataSource(data, pageSize,
sortColumn, desc)
- 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 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] = {
+ 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(batch: BatchUIData): Seq[Node] = {
+ val batchTime = batch.batchTime.milliseconds
+ val formattedBatchTime = UIUtils.formatBatchTime(batchTime, batchInterval)
+ val numRecords = batch.numRecords
+ val schedulingDelay = batch.schedulingDelay
+ val formattedSchedulingDelay =
schedulingDelay.map(SparkUIUtils.formatDuration).getOrElse("-")
+ val processingTime = batch.processingDelay
+ val formattedProcessingTime =
processingTime.map(SparkUIUtils.formatDuration).getOrElse("-")
+ val batchTimeId = s"batch-$batchTime"
val totalDelay = batch.totalDelay
val formattedTotalDelay =
totalDelay.map(SparkUIUtils.formatDuration).getOrElse("-")
- baseRow(batch) ++ {
- <td sorttable_customkey={totalDelay.getOrElse(Long.MaxValue).toString}>
+ <tr>
+ <td id={batchTimeId}>
+ <a href={s"batch?id=$batchTime"}>
+ {formattedBatchTime}
+ </a>
+ </td>
+ <td>
+ {numRecords.toString} records
+ </td>
+ <td>
+ {formattedSchedulingDelay}
+ </td>
+ <td>
+ {formattedProcessingTime}
+ </td>
+ <td>
{formattedTotalDelay}
</td>
- } ++ createOutputOperationProgressBar(batch)++ {
- if (firstFailureReason.nonEmpty) {
- getFirstFailureTableCell(batch)
- } else {
- Nil
+ <td class="progress-cell">
+ {SparkUIUtils.makeProgressBar(started = batch.numActiveOutputOp,
+ completed = batch.numCompletedOutputOp, failed =
batch.numFailedOutputOp, skipped = 0,
Review comment:
I think the same code is there in the method
`createOutputOperationProgressBar`? Can't we use that?
----------------------------------------------------------------
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]
