Revert "[SPARK-31953][SS] Add Spark Structured Streaming History Server Support"

This reverts commit 4f96670.

Author: HyukjinKwon

dev/.rat-excludes

@@ -123,7 +123,6 @@ SessionHandler.java
 GangliaReporter.java
 application_1578436911597_0052
 config.properties
-local-1596020211915
 app-20200706201101-0003
 py.typed
 _metadata

sql/core/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin

@@ -1,2 +1 @@
 org.apache.spark.sql.execution.ui.SQLHistoryServerPlugin
-org.apache.spark.sql.execution.ui.StreamingQueryHistoryServerPlugin

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala

@@ -31,21 +31,16 @@ import org.apache.spark.util.ListenerBus
  * Spark listener bus, so that it can receive [[StreamingQueryListener.Event]]s and dispatch them
  * to StreamingQueryListeners.
  *
- * Note 1: Each bus and its registered listeners are associated with a single SparkSession
+ * Note that each bus and its registered listeners are associated with a single SparkSession
  * and StreamingQueryManager. So this bus will dispatch events to registered listeners for only
  * those queries that were started in the associated SparkSession.
- *
- * Note 2: To rebuild Structured Streaming UI in SHS, this bus will be registered into
- * [[org.apache.spark.scheduler.ReplayListenerBus]]. We check `sparkListenerBus` defined or not to
- * determine how to process [[StreamingQueryListener.Event]]. If false, it means this bus is used to
- * replay all streaming query event from eventLog.
  */
-class StreamingQueryListenerBus(sparkListenerBus: Option[LiveListenerBus])
+class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus)
   extends SparkListener with ListenerBus[StreamingQueryListener, StreamingQueryListener.Event] {
 
   import StreamingQueryListener._
 
-  sparkListenerBus.foreach(_.addToQueue(this, StreamingQueryListenerBus.STREAM_EVENT_QUERY))
+  sparkListenerBus.addToQueue(this, StreamingQueryListenerBus.STREAM_EVENT_QUERY)
 
   /**
    * RunIds of active queries whose events are supposed to be forwarded by this ListenerBus
@@ -72,11 +67,11 @@ class StreamingQueryListenerBus(sparkListenerBus: Option[LiveListenerBus])
     event match {
       case s: QueryStartedEvent =>
         activeQueryRunIds.synchronized { activeQueryRunIds += s.runId }
-        sparkListenerBus.foreach(bus => bus.post(s))
+        sparkListenerBus.post(s)
         // post to local listeners to trigger callbacks
         postToAll(s)
       case _ =>
-        sparkListenerBus.foreach(bus => bus.post(event))
+        sparkListenerBus.post(event)
     }
   }
 
@@ -100,11 +95,7 @@ class StreamingQueryListenerBus(sparkListenerBus: Option[LiveListenerBus])
         // synchronously and the ones attached to LiveListenerBus asynchronously. Therefore,
         // we need to ignore QueryStartedEvent if this method is called within SparkListenerBus
         // thread
-        //
-        // When loaded by Spark History Server, we should process all event coming from replay
-        // listener bus.
-        if (sparkListenerBus.isEmpty || !LiveListenerBus.withinListenerThread.value ||
-            !e.isInstanceOf[QueryStartedEvent])  {
+        if (!LiveListenerBus.withinListenerThread.value || !e.isInstanceOf[QueryStartedEvent]) {
           postToAll(e)
         }
       case _ =>
@@ -119,10 +110,7 @@ class StreamingQueryListenerBus(sparkListenerBus: Option[LiveListenerBus])
       listener: StreamingQueryListener,
       event: StreamingQueryListener.Event): Unit = {
     def shouldReport(runId: UUID): Boolean = {
-      // When loaded by Spark History Server, we should process all event coming from replay
-      // listener bus.
-      sparkListenerBus.isEmpty ||
-        activeQueryRunIds.synchronized { activeQueryRunIds.contains(runId) }
+      activeQueryRunIds.synchronized { activeQueryRunIds.contains(runId) }
     }
 
     event match {

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/StreamingQueryHistoryServerPlugin.scala

@@ -34,7 +34,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.execution.CacheManager
 import org.apache.spark.sql.execution.streaming.StreamExecution
-import org.apache.spark.sql.execution.ui.{SQLAppStatusListener, SQLAppStatusStore, SQLTab, StreamingQueryStatusStore}
+import org.apache.spark.sql.execution.ui.{SQLAppStatusListener, SQLAppStatusStore, SQLTab}
 import org.apache.spark.sql.internal.StaticSQLConf._
 import org.apache.spark.sql.streaming.ui.{StreamingQueryStatusListener, StreamingQueryTab}
 import org.apache.spark.status.ElementTrackingStore
@@ -111,9 +111,9 @@ private[sql] class SharedState(
   lazy val streamingQueryStatusListener: Option[StreamingQueryStatusListener] = {
     sparkContext.ui.flatMap { ui =>
       if (conf.get(STREAMING_UI_ENABLED)) {
-        val kvStore = sparkContext.statusStore.store.asInstanceOf[ElementTrackingStore]
-        new StreamingQueryTab(new StreamingQueryStatusStore(kvStore), ui)
-        Some(new StreamingQueryStatusListener(conf, kvStore))
+        val statusListener = new StreamingQueryStatusListener(conf)
+        new StreamingQueryTab(statusListener, ui)
+        Some(statusListener)
       } else {
         None
       }

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/StreamingQueryStatusStore.scala

@@ -49,8 +49,7 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession) extends Lo
 
   private[sql] val stateStoreCoordinator =
     StateStoreCoordinatorRef.forDriver(sparkSession.sparkContext.env)
-  private val listenerBus =
-    new StreamingQueryListenerBus(Some(sparkSession.sparkContext.listenerBus))
+  private val listenerBus = new StreamingQueryListenerBus(sparkSession.sparkContext.listenerBus)
 
   @GuardedBy("activeQueriesSharedLock")
   private val activeQueries = new mutable.HashMap[UUID, StreamingQuery]

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

@@ -40,8 +40,8 @@ private[ui] class StreamingQueryPage(parent: StreamingQueryTab)
   }
 
   private def generateStreamingQueryTable(request: HttpServletRequest): Seq[Node] = {
-    val (activeQueries, inactiveQueries) =
-      parent.store.allQueryUIData.partition(_.summary.isActive)
+    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
@@ -176,7 +176,7 @@ class StreamingQueryPagedTable(
     val streamingQuery = query.streamingUIData
     val statisticsLink = "%s/%s/statistics?id=%s"
       .format(SparkUIUtils.prependBaseUri(request, parent.basePath), parent.prefix,
-        streamingQuery.summary.runId)
+        streamingQuery.runId)
 
     def details(detail: Any): Seq[Node] = {
       if (isActive) {
@@ -194,14 +194,14 @@ class StreamingQueryPagedTable(
     <tr>
       <td>{UIUtils.getQueryName(streamingQuery)}</td>
       <td>{UIUtils.getQueryStatus(streamingQuery)}</td>
-      <td>{streamingQuery.summary.id}</td>
-      <td><a href={statisticsLink}>{streamingQuery.summary.runId}</a></td>
-      <td>{SparkUIUtils.formatDate(streamingQuery.summary.startTimestamp)}</td>
+      <td>{streamingQuery.id}</td>
+      <td><a href={statisticsLink}>{streamingQuery.runId}</a></td>
+      <td>{SparkUIUtils.formatDate(streamingQuery.startTimestamp)}</td>
       <td>{SparkUIUtils.formatDurationVerbose(query.duration)}</td>
       <td>{withNoProgress(streamingQuery, {query.avgInput.formatted("%.2f")}, "NaN")}</td>
       <td>{withNoProgress(streamingQuery, {query.avgProcess.formatted("%.2f")}, "NaN")}</td>
       <td>{withNoProgress(streamingQuery, {streamingQuery.lastProgress.batchId}, "NaN")}</td>
-      {details(streamingQuery.summary.exception.getOrElse("-"))}
+      {details(streamingQuery.exception.getOrElse("-"))}
     </tr>
   }
 }
@@ -222,32 +222,32 @@ class StreamingQueryDataSource(uiData: Seq[StreamingQueryUIData], sortColumn: St
 
   override def sliceData(from: Int, to: Int): Seq[StructuredStreamingRow] = data.slice(from, to)
 
-  private def streamingRow(uiData: StreamingQueryUIData): StructuredStreamingRow = {
+  private def streamingRow(query: StreamingQueryUIData): StructuredStreamingRow = {
     val duration = if (isActive) {
-      System.currentTimeMillis() - uiData.summary.startTimestamp
+      System.currentTimeMillis() - query.startTimestamp
     } else {
-      withNoProgress(uiData, {
-        val endTimeMs = uiData.lastProgress.timestamp
-        parseProgressTimestamp(endTimeMs) - uiData.summary.startTimestamp
+      withNoProgress(query, {
+        val endTimeMs = query.lastProgress.timestamp
+        parseProgressTimestamp(endTimeMs) - query.startTimestamp
       }, 0)
     }
 
-    val avgInput = (uiData.recentProgress.map(p => withNumberInvalid(p.inputRowsPerSecond)).sum /
-      uiData.recentProgress.length)
+    val avgInput = (query.recentProgress.map(p => withNumberInvalid(p.inputRowsPerSecond)).sum /
+      query.recentProgress.length)
 
-    val avgProcess = (uiData.recentProgress.map(p =>
-      withNumberInvalid(p.processedRowsPerSecond)).sum / uiData.recentProgress.length)
+    val avgProcess = (query.recentProgress.map(p =>
+      withNumberInvalid(p.processedRowsPerSecond)).sum / query.recentProgress.length)
 
-    StructuredStreamingRow(duration, avgInput, avgProcess, uiData)
+    StructuredStreamingRow(duration, avgInput, avgProcess, query)
   }
 
   private def ordering(sortColumn: String, desc: Boolean): Ordering[StructuredStreamingRow] = {
     val ordering: Ordering[StructuredStreamingRow] = sortColumn match {
-      case "Name" => Ordering.by(row => UIUtils.getQueryName(row.streamingUIData))
-      case "Status" => Ordering.by(row => UIUtils.getQueryStatus(row.streamingUIData))
-      case "ID" => Ordering.by(_.streamingUIData.summary.id)
-      case "Run ID" => Ordering.by(_.streamingUIData.summary.runId)
-      case "Start Time" => Ordering.by(_.streamingUIData.summary.startTimestamp)
+      case "Name" => Ordering.by(q => UIUtils.getQueryName(q.streamingUIData))
+      case "Status" => Ordering.by(q => UIUtils.getQueryStatus(q.streamingUIData))
+      case "ID" => Ordering.by(_.streamingUIData.id)
+      case "Run ID" => Ordering.by(_.streamingUIData.runId)
+      case "Start Time" => Ordering.by(_.streamingUIData.startTimestamp)
       case "Duration" => Ordering.by(_.duration)
       case "Avg Input /sec" => Ordering.by(_.avgInput)
       case "Avg Process /sec" => Ordering.by(_.avgProcess)

sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala

@@ -58,8 +58,8 @@ private[ui] class StreamingQueryStatisticsPage(parent: StreamingQueryTab)
     val parameterId = request.getParameter("id")
     require(parameterId != null && parameterId.nonEmpty, "Missing id parameter")
 
-    val query = parent.store.allQueryUIData.find { uiData =>
-      uiData.summary.runId.equals(UUID.fromString(parameterId))
+    val query = parent.statusListener.allQueryStatus.find { case q =>
+      q.runId.equals(UUID.fromString(parameterId))
     }.getOrElse(throw new IllegalArgumentException(s"Failed to find streaming query $parameterId"))
 
     val resources = generateLoadResources(request)
@@ -109,35 +109,34 @@ private[ui] class StreamingQueryStatisticsPage(parent: StreamingQueryTab)
     <script>{Unparsed(js)}</script>
   }
 
-  def generateBasicInfo(uiData: StreamingQueryUIData): Seq[Node] = {
-    val duration = if (uiData.summary.isActive) {
-      val durationMs = System.currentTimeMillis() - uiData.summary.startTimestamp
-      SparkUIUtils.formatDurationVerbose(durationMs)
+  def generateBasicInfo(query: StreamingQueryUIData): Seq[Node] = {
+    val duration = if (query.isActive) {
+      SparkUIUtils.formatDurationVerbose(System.currentTimeMillis() - query.startTimestamp)
     } else {
-      withNoProgress(uiData, {
-        val end = uiData.lastProgress.timestamp
-        val start = uiData.recentProgress.head.timestamp
+      withNoProgress(query, {
+        val end = query.lastProgress.timestamp
+        val start = query.recentProgress.head.timestamp
         SparkUIUtils.formatDurationVerbose(
           parseProgressTimestamp(end) - parseProgressTimestamp(start))
       }, "-")
     }
 
-    val name = UIUtils.getQueryName(uiData)
-    val numBatches = withNoProgress(uiData, { uiData.lastProgress.batchId + 1L }, 0)
+    val name = UIUtils.getQueryName(query)
+    val numBatches = withNoProgress(query, { query.lastProgress.batchId + 1L }, 0)
     <div>Running batches for
       <strong>
         {duration}
       </strong>
       since
       <strong>
-        {SparkUIUtils.formatDate(uiData.summary.startTimestamp)}
+        {SparkUIUtils.formatDate(query.startTimestamp)}
       </strong>
       (<strong>{numBatches}</strong> completed batches)
     </div>
     <br />
     <div><strong>Name: </strong>{name}</div>
-    <div><strong>Id: </strong>{uiData.summary.id}</div>
-    <div><strong>RunId: </strong>{uiData.summary.runId}</div>
+    <div><strong>Id: </strong>{query.id}</div>
+    <div><strong>RunId: </strong>{query.runId}</div>
     <br />
   }