iRakson commented on a change in pull request #28485:
URL: https://github.com/apache/spark/pull/28485#discussion_r429645268
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryPage.scala
##########
@@ -35,11 +39,147 @@ private[ui] class StreamingQueryPage(parent:
StreamingQueryTab)
SparkUIUtils.headerSparkPage(request, "Streaming Query", content, parent)
}
- def generateDataRow(request: HttpServletRequest, queryActive: Boolean)
- (query: StreamingQueryUIData): Seq[Node] = {
+ private def generateStreamingQueryTable(request: HttpServletRequest):
Seq[Node] = {
+ val (activeQueries, inactiveQueries) = parent.statusListener.allQueryStatus
+ .partition(_.isActive)
+
+ val content = mutable.ListBuffer[Node]()
+ // show active queries table only if there is at least one active query
+ if (activeQueries.nonEmpty) {
+ // scalastyle:off
+ content ++=
+ <span id="active" class="collapse-aggregated-activeQueries
collapse-table"
+
onClick="collapseTable('collapse-aggregated-activeQueries','aggregated-activeQueries')">
+ <h5 id="activequeries">
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a>Active Streaming Queries ({activeQueries.length})</a>
+ </h5>
+ </span> ++
+ <div>
+ <ul class="aggregated-activeQueries collapsible-table">
+ {queryTable(activeQueries, request, "active")}
+ </ul>
+ </div>
+ // scalastyle:on
+ }
+ // show active queries table only if there is at least one completed query
+ if (inactiveQueries.nonEmpty) {
+ // scalastyle:off
+ content ++=
+ <span id="completed" class="collapse-aggregated-completedQueries
collapse-table"
+
onClick="collapseTable('collapse-aggregated-completedQueries','aggregated-completedQueries')">
+ <h5 id="completedqueries">
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a>Completed Streaming Queries ({inactiveQueries.length})</a>
+ </h5>
+ </span> ++
+ <div>
+ <ul class="aggregated-completedQueries collapsible-table">
+ {queryTable(inactiveQueries, request, "completed")}
+ </ul>
+ </div>
+ // scalastyle:on
+ }
+ content
+ }
+
+ private def queryTable(data: Seq[StreamingQueryUIData], request:
HttpServletRequest,
+ tableTag: String): Seq[Node] = {
+
+ val isActive = if (tableTag.contains("active")) true else false
+ val page =
Option(request.getParameter(s"$tableTag.page")).map(_.toInt).getOrElse(1)
+
+ try {
+ new StreamingQueryPagedTable(
+ request,
+ parent,
+ data,
+ tableTag,
+ isActive,
+ SparkUIUtils.prependBaseUri(request, parent.basePath),
+ "StreamingQuery"
+ ).table(page)
+ } catch {
+ case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+ }
+}
+
+class StreamingQueryPagedTable(
+ request: HttpServletRequest,
+ parent: StreamingQueryTab,
+ data: Seq[StreamingQueryUIData],
+ tableTag: String,
+ isActive: Boolean,
+ basePath: String,
+ subPath: String) extends PagedTable[StructuredStreamingRow] {
+
+ private val (sortColumn, sortDesc, pageSize) = getTableParameters(request,
tableTag, "Start Time")
+ private val parameterPath =
s"$basePath/$subPath/?${getParameterOtherTable(request, tableTag)}"
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override def tableId: String = s"$tableTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable
table-cell-width-limited"
+
+ override def pageSizeFormField: String = s"$tableTag.pageSize"
+
+ override def pageNumberFormField: String = s"$tableTag.page"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$tableTag.sort=$encodedSortColumn" +
+ s"&$tableTag.desc=$sortDesc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableTag"
+ }
+
+ override def goButtonFormPath: String =
+
s"$parameterPath&$tableTag.sort=$encodedSortColumn&$tableTag.desc=$sortDesc#$tableTag"
+
+ override def dataSource: PagedDataSource[StructuredStreamingRow] =
+ new StreamingQueryDataSource(data, sortColumn, sortDesc, pageSize,
isActive)
+
+ override def headers: Seq[Node] = {
+ val headerAndCss: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("Name", true, None),
+ ("Status", false, None),
+ ("ID", true, None),
+ ("Run ID", true, None),
+ ("Start Time", true, None),
+ ("Duration", true, None),
+ ("Avg Input /sec", true, None),
Review comment:
@srowen @uncleGen @sarutak Should we add tooltips for these columns.
----------------------------------------------------------------
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
