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_r358001626
##########
File path:
streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
##########
@@ -482,11 +484,61 @@ private[ui] class StreamingPage(parent: StreamingTab)
</tr>
}
- private def generateBatchListTables(): Seq[Node] = {
+ private def streamingBatchTable(
+ request: HttpServletRequest,
+ batchData: Seq[BatchUIData],
+ streamingBatchTag: String,
+ batchInterval: Long): Seq[Node] = {
+ val parameterOtherTable = request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(streamingBatchTag))
+ .map { case (name, vals) =>
+ name + "=" + vals(0)
+ }
+
+ val parameterStreamingBatchPage = request.getParameter(streamingBatchTag +
".page")
+ val parameterStreamingBatchSortColumn =
request.getParameter(streamingBatchTag + ".sort")
+ val parameterStreamingBatchSortDesc =
request.getParameter(streamingBatchTag + ".desc")
+ val parameterStreamingBatchPageSize =
request.getParameter(streamingBatchTag + ".pageSize")
+ val streamingBatchPage =
Option(parameterStreamingBatchPage).map(_.toInt).getOrElse(1)
+ val streamingBatchSortColumn =
Option(parameterStreamingBatchSortColumn).map { sortColumn =>
+ SparkUIUtils.decodeURLParameter(sortColumn)
+ }.getOrElse("Batch Time")
+ val streamingBatchSortDesc =
Option(parameterStreamingBatchSortDesc).map(_.toBoolean).getOrElse(
+ streamingBatchSortColumn == "Batch Time"
+ )
+ val streamingBatchPageSize =
Option(parameterStreamingBatchPageSize).map(_.toInt).getOrElse(100)
+
+ try {
+ new StreamingBatchPagedTable(
+ request,
+ parent,
+ batchInterval,
+ batchData,
+ streamingBatchTag,
+ SparkUIUtils.prependBaseUri(request, parent.basePath),
+ "streaming", // subPath
+ parameterOtherTable,
+ pageSize = streamingBatchPageSize,
+ sortColumn = streamingBatchSortColumn,
+ desc = streamingBatchSortDesc
+ ).table(streamingBatchPage)
+ } catch {
+ case e @ (_ : IllegalArgumentException | _ : IndexOutOfBoundsException)
=>
+ <div class="alert alert-error">
+ <p>Error while rendering job table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+ }
+
+ private def generateBatchListTables(request: HttpServletRequest): Seq[Node]
= {
val runningBatches =
listener.runningBatches.sortBy(_.batchTime.milliseconds).reverse
val waitingBatches =
listener.waitingBatches.sortBy(_.batchTime.milliseconds).reverse
val completedBatches = listener.retainedCompletedBatches.
sortBy(_.batchTime.milliseconds).reverse
+ val activeBatchData = waitingBatches ++ runningBatches
Review comment:
Can we simply append the two table data? Could you please check the output
of the pages before and after the PR, which contains both runningBatches and
waitingBatches?
----------------------------------------------------------------
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]
