HeartSaVioR commented on a change in pull request #30151:
URL: https://github.com/apache/spark/pull/30151#discussion_r515757520
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
Review comment:
nit: we tend to append this line to the last line of the parameter if
the last line of the parameter is not long.
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
Review comment:
Let's add `state` in ID of div tag explicitly, so that these can be
clearly separated with other div tags.
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-updated-rows-histogram"}>{graphUIDataForNumberUpdatedRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated State Memory Used In Bytes
{SparkUIUtils.tooltip("Aggregated state memory used in bytes.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-memory-used-bytes-timeline"}>{graphUIDataForMemoryUsedBytes.generateTimelineHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-updated-rows-histogram"}>{graphUIDataForNumberUpdatedRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated State Memory Used In Bytes
{SparkUIUtils.tooltip("Aggregated state memory used in bytes.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-memory-used-bytes-timeline"}>{graphUIDataForMemoryUsedBytes.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-memory-used-bytes-histogram"}>{graphUIDataForMemoryUsedBytes.generateHistogramHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-updated-rows-histogram"}>{graphUIDataForNumberUpdatedRows.generateHistogramHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-updated-rows-histogram"}>{graphUIDataForNumberUpdatedRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated State Memory Used In Bytes
{SparkUIUtils.tooltip("Aggregated state memory used in bytes.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-memory-used-bytes-timeline"}>{graphUIDataForMemoryUsedBytes.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-memory-used-bytes-histogram"}>{graphUIDataForMemoryUsedBytes.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of State Rows Dropped By
Watermark {SparkUIUtils.tooltip("Aggregated number of state rows dropped by
watermark.", "right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-rows-dropped-by-watermark-timeline"}>{graphUIDataForNumRowsDroppedByWatermark.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-rows-dropped-by-watermark-histogram"}>{graphUIDataForNumRowsDroppedByWatermark.generateHistogramHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-total-rows-histogram"}>{graphUIDataForNumberTotalRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Updated State Rows
{SparkUIUtils.tooltip("Aggregated number of updated state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-updated-rows-timeline"}>{graphUIDataForNumberUpdatedRows.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-num-updated-rows-histogram"}>{graphUIDataForNumberUpdatedRows.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated State Memory Used In Bytes
{SparkUIUtils.tooltip("Aggregated state memory used in bytes.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-memory-used-bytes-timeline"}>{graphUIDataForMemoryUsedBytes.generateTimelineHtml(jsCollector)}</td>
+ <td
class={"aggregated-memory-used-bytes-histogram"}>{graphUIDataForMemoryUsedBytes.generateHistogramHtml(jsCollector)}</td>
+ </tr>
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of State Rows Dropped By
Watermark {SparkUIUtils.tooltip("Aggregated number of state rows dropped by
watermark.", "right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-rows-dropped-by-watermark-timeline"}>{graphUIDataForNumRowsDroppedByWatermark.generateTimelineHtml(jsCollector)}</td>
Review comment:
ditto
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatisticsPage.scala
##########
@@ -126,6 +126,123 @@ private[ui] class StreamingQueryStatisticsPage(parent:
StreamingQueryTab)
<br />
}
+ def generateAggregatedStateOperators(
+ query: StreamingQueryUIData,
+ minBatchTime: Long,
+ maxBatchTime: Long,
+ jsCollector: JsCollector
+ ): NodeBuffer = {
+ // This is made sure on caller side but put it here to be defensive
+ require(query.lastProgress != null)
+ if (query.lastProgress.stateOperators.nonEmpty) {
+ val numRowsTotalData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsTotal).sum.toDouble))
+ val maxNumRowsTotal = numRowsTotalData.maxBy(_._2)._2
+
+ val numRowsUpdatedData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsUpdated).sum.toDouble))
+ val maxNumRowsUpdated = numRowsUpdatedData.maxBy(_._2)._2
+
+ val memoryUsedBytesData = query.recentProgress.map(p =>
(parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.memoryUsedBytes).sum.toDouble))
+ val maxMemoryUsedBytes = memoryUsedBytesData.maxBy(_._2)._2
+
+ val numRowsDroppedByWatermarkData = query.recentProgress
+ .map(p => (parseProgressTimestamp(p.timestamp),
+ p.stateOperators.map(_.numRowsDroppedByWatermark).sum.toDouble))
+ val maxNumRowsDroppedByWatermark =
numRowsDroppedByWatermarkData.maxBy(_._2)._2
+
+ val graphUIDataForNumberTotalRows =
+ new GraphUIData(
+ "aggregated-num-total-rows-timeline",
+ "aggregated-num-total-rows-histogram",
+ numRowsTotalData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsTotal,
+ "records")
+ graphUIDataForNumberTotalRows.generateDataJs(jsCollector)
+
+ val graphUIDataForNumberUpdatedRows =
+ new GraphUIData(
+ "aggregated-num-updated-rows-timeline",
+ "aggregated-num-updated-rows-histogram",
+ numRowsUpdatedData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsUpdated,
+ "records")
+ graphUIDataForNumberUpdatedRows.generateDataJs(jsCollector)
+
+ val graphUIDataForMemoryUsedBytes =
+ new GraphUIData(
+ "aggregated-memory-used-bytes-timeline",
+ "aggregated-memory-used-bytes-histogram",
+ memoryUsedBytesData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxMemoryUsedBytes,
+ "bytes")
+ graphUIDataForMemoryUsedBytes.generateDataJs(jsCollector)
+
+ val graphUIDataForNumRowsDroppedByWatermark =
+ new GraphUIData(
+ "aggregated-num-rows-dropped-by-watermark-timeline",
+ "aggregated-num-rows-dropped-by-watermark-histogram",
+ numRowsDroppedByWatermarkData,
+ minBatchTime,
+ maxBatchTime,
+ 0,
+ maxNumRowsDroppedByWatermark,
+ "records")
+ graphUIDataForNumRowsDroppedByWatermark.generateDataJs(jsCollector)
+
+ // scalastyle:off
+ <tr>
+ <td style="vertical-align: middle;">
+ <div style="width: 160px;">
+ <div><strong>Aggregated Number Of Total State Rows
{SparkUIUtils.tooltip("Aggregated number of total state rows.",
"right")}</strong></div>
+ </div>
+ </td>
+ <td
class={"aggregated-num-total-rows-timeline"}>{graphUIDataForNumberTotalRows.generateTimelineHtml(jsCollector)}</td>
Review comment:
Also, is wrapping `{` and `}` required? I see other classes in this file
don't, hence the question.
----------------------------------------------------------------
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]
