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_r358072287
##########
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:
I am not sure we can simply append the table. Please refer earlier code. To
check, if there is no change, could you please attach screenshot for that
(before and after PR)
https://github.com/apache/spark/blob/1fc353d51a62cb554e6af23dbc9a613e214e3af1/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala#L130-L135
----------------------------------------------------------------
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]
