You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "bogao007 (via GitHub)" <gi...@apache.org> on 2023/04/21 06:04:06 UTC
[GitHub] [spark] bogao007 opened a new pull request, #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
bogao007 opened a new pull request, #40895:
URL: https://github.com/apache/spark/pull/40895
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Implemented StreamingQueryProgress for Spark Connect. Also added related structs under the legacy `StreamingQueryProgress`.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Since Spark Connect transfers streaming progress as full “json”, implemented `fromJson` method to deserialize the json string to the legacy `StreamingQueryProgress`.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added new unit test suite.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173336296
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,265 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger.
+ * @param json A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark](
Review Comment:
Mostly copied from `org.apache.spark.sql.streaming.progress.scala`, did some small adjustment.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173414989
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryProgressSuite.scala:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import org.apache.spark.sql.SQLHelper
+import org.apache.spark.sql.connect.client.util.RemoteSparkSession
+
+class StreamingQueryProgressSuite extends RemoteSparkSession with SQLHelper {
+
+ test("test from json") {
+ val jsonString =
+ """{
+ | "id": "52de8a4e-1af7-4358-83e1-74d5d44bbc1b",
+ | "runId": "46680bd2-bb67-4828-90d6-3f76f43acd70",
+ | "name": "myName",
+ | "timestamp": "2023-04-20T04:52:47.890Z",
+ | "batchId": 169095,
+ | "numInputRows": 839,
+ | "inputRowsPerSecond": 0.0,
+ | "processedRowsPerSecond": 252.33082706766916,
+ | "durationMs":
+ | {
+ | "addBatch": 2240,
+ | "getBatch": 99,
+ | "latestOffset": 74,
+ | "queryPlanning": 3,
+ | "triggerExecution": 3325,
+ | "walCommit": 325
+ | },
+ | "eventTime" : {
+ | "avg" : "2016-12-05T20:54:20.827Z",
+ | "max" : "2016-12-05T20:54:20.827Z",
+ | "min" : "2016-12-05T20:54:20.827Z",
+ | "watermark" : "2016-12-05T20:54:20.827Z"
+ | },
+ | "stateOperators" : [ {
+ | "operatorName" : "op1",
+ | "numRowsTotal" : 0,
+ | "numRowsUpdated" : 1,
+ | "allUpdatesTimeMs" : 1,
+ | "numRowsRemoved" : 2,
+ | "allRemovalsTimeMs" : 34,
+ | "commitTimeMs" : 23,
+ | "memoryUsedBytes" : 3,
+ | "numRowsDroppedByWatermark" : 0,
+ | "numShufflePartitions" : 2,
+ | "numStateStoreInstances" : 2,
+ | "customMetrics" : {
+ | "loadedMapCacheHitCount" : 1,
+ | "loadedMapCacheMissCount" : 0,
+ | "stateOnCurrentVersionSizeBytes" : 2
+ | }
+ | } ],
+ | "sources" : [ {
+ | "description" : "source",
+ | "startOffset" : "123",
+ | "endOffset" : "456",
+ | "latestOffset" : "789",
+ | "numInputRows" : 678,
+ | "inputRowsPerSecond" : 10.0
+ | } ],
+ | "sink" : {
+ | "description" : "sink",
+ | "numOutputRows" : -1
+ | }
+ |}""".stripMargin
+
+ val streamingQueryProgress = new StreamingQueryProgress(jsonString)
+ val result = streamingQueryProgress.fromJson()
+ assert("52de8a4e-1af7-4358-83e1-74d5d44bbc1b" === result.id.toString)
+ assert("46680bd2-bb67-4828-90d6-3f76f43acd70" === result.runId.toString)
+ assert(839 === result.numInputRows)
Review Comment:
hmm... there is only one `SourceProgress` instance in json, so this should be 678?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173337245
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,265 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger.
+ * @param json A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark](
+ val operatorName: String,
+ val numRowsTotal: Long,
+ val numRowsUpdated: Long,
+ val allUpdatesTimeMs: Long,
+ val numRowsRemoved: Long,
+ val allRemovalsTimeMs: Long,
+ val commitTimeMs: Long,
+ val memoryUsedBytes: Long,
+ val numRowsDroppedByWatermark: Long,
+ val numShufflePartitions: Long,
+ val numStateStoreInstances: Long,
+ val customMetrics: ju.Map[String, JLong] = new ju.HashMap()
+ ) extends Serializable {
+
+ /** The compact JSON representation of this progress. */
+ def json: String = compact(render(jsonValue))
+
+ /** The pretty (i.e. indented) JSON representation of this progress. */
+ def prettyJson: String = pretty(render(jsonValue))
+
+ private[sql] def jsonValue: JValue = {
+ ("operatorName" -> JString(operatorName)) ~
+ ("numRowsTotal" -> JInt(numRowsTotal)) ~
+ ("numRowsUpdated" -> JInt(numRowsUpdated)) ~
+ ("allUpdatesTimeMs" -> JInt(allUpdatesTimeMs)) ~
+ ("numRowsRemoved" -> JInt(numRowsRemoved)) ~
+ ("allRemovalsTimeMs" -> JInt(allRemovalsTimeMs)) ~
+ ("commitTimeMs" -> JInt(commitTimeMs)) ~
+ ("memoryUsedBytes" -> JInt(memoryUsedBytes)) ~
+ ("numRowsDroppedByWatermark" -> JInt(numRowsDroppedByWatermark)) ~
+ ("numShufflePartitions" -> JInt(numShufflePartitions)) ~
+ ("numStateStoreInstances" -> JInt(numStateStoreInstances)) ~
+ ("customMetrics" -> {
+ if (!customMetrics.isEmpty) {
+ val keys = customMetrics.keySet.asScala.toSeq.sorted
+ keys.map { k => k -> JInt(customMetrics.get(k).toLong) : JObject }.reduce(_ ~ _)
+ } else {
+ JNothing
+ }
+ })
+ }
+
+ override def toString: String = prettyJson
+}
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger. Each event relates to processing done for a single trigger of the streaming
+ * query. Events are emitted even when no new data is available to be processed.
+ *
+ * @param id A unique query id that persists across restarts. See `StreamingQuery.id()`.
+ * @param runId A query id that is unique for every start/restart. See `StreamingQuery.runId()`.
+ * @param name User-specified name of the query, null if not specified.
+ * @param timestamp Beginning time of the trigger in ISO8601 format, i.e. UTC timestamps.
+ * @param batchId A unique id for the current batch of data being processed. Note that in the
+ * case of retries after a failure a given batchId my be executed more than once.
+ * Similarly, when there is no data to be processed, the batchId will not be
+ * incremented.
+ * @param numInputRows The total number of records read from all the sources.
+ * @param inputRowsPerSecond The total rate at which data is arriving from all the sources.
+ * @param processedRowsPerSecond The total rate at which data from all the sources is being
+ * processed by Spark.
+ * @param batchDuration The process duration of each batch.
+ * @param durationMs The amount of time taken to perform various operations in milliseconds.
+ * @param eventTime Statistics of event time seen in this batch. It may contain the following keys:
+ * {{{
+ * "max" -> "2016-12-05T20:54:20.827Z" // maximum event time seen in this trigger
+ * "min" -> "2016-12-05T20:54:20.827Z" // minimum event time seen in this trigger
+ * "avg" -> "2016-12-05T20:54:20.827Z" // average event time seen in this trigger
+ * "watermark" -> "2016-12-05T20:54:20.827Z" // watermark used in this trigger
+ * }}}
+ * All timestamps are in ISO8601 format, i.e. UTC timestamps.
+ * @param stateOperators Information about operators in the query that store state.
+ * @param sources detailed statistics on data being read from each of the streaming sources.
+ * @since 2.1.0
+ */
+@Evolving
+class LegacyStreamingQueryProgress private[spark](
+ val id: UUID,
+ val runId: UUID,
+ val name: String,
+ val timestamp: String,
+ val batchId: Long,
+ val numInputRows: Long,
+ val inputRowsPerSecond: Double,
+ val processedRowsPerSecond: Double,
+ val batchDuration: Long,
+ val durationMs: ju.Map[String, JLong],
+ val eventTime: ju.Map[String, String],
+ val stateOperators: Array[StateOperatorProgress],
+ val sources: Array[SourceProgress],
+ val sink: SinkProgress) extends Serializable {
Review Comment:
I removed `observedMetrics` from the original class since it has some issues to deserialize the `GenericRowWithSchema` type. If we don't need this then it should be okay, otherwise I'll need to investigate this deserialization issue.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173337245
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,265 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger.
+ * @param json A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark](
+ val operatorName: String,
+ val numRowsTotal: Long,
+ val numRowsUpdated: Long,
+ val allUpdatesTimeMs: Long,
+ val numRowsRemoved: Long,
+ val allRemovalsTimeMs: Long,
+ val commitTimeMs: Long,
+ val memoryUsedBytes: Long,
+ val numRowsDroppedByWatermark: Long,
+ val numShufflePartitions: Long,
+ val numStateStoreInstances: Long,
+ val customMetrics: ju.Map[String, JLong] = new ju.HashMap()
+ ) extends Serializable {
+
+ /** The compact JSON representation of this progress. */
+ def json: String = compact(render(jsonValue))
+
+ /** The pretty (i.e. indented) JSON representation of this progress. */
+ def prettyJson: String = pretty(render(jsonValue))
+
+ private[sql] def jsonValue: JValue = {
+ ("operatorName" -> JString(operatorName)) ~
+ ("numRowsTotal" -> JInt(numRowsTotal)) ~
+ ("numRowsUpdated" -> JInt(numRowsUpdated)) ~
+ ("allUpdatesTimeMs" -> JInt(allUpdatesTimeMs)) ~
+ ("numRowsRemoved" -> JInt(numRowsRemoved)) ~
+ ("allRemovalsTimeMs" -> JInt(allRemovalsTimeMs)) ~
+ ("commitTimeMs" -> JInt(commitTimeMs)) ~
+ ("memoryUsedBytes" -> JInt(memoryUsedBytes)) ~
+ ("numRowsDroppedByWatermark" -> JInt(numRowsDroppedByWatermark)) ~
+ ("numShufflePartitions" -> JInt(numShufflePartitions)) ~
+ ("numStateStoreInstances" -> JInt(numStateStoreInstances)) ~
+ ("customMetrics" -> {
+ if (!customMetrics.isEmpty) {
+ val keys = customMetrics.keySet.asScala.toSeq.sorted
+ keys.map { k => k -> JInt(customMetrics.get(k).toLong) : JObject }.reduce(_ ~ _)
+ } else {
+ JNothing
+ }
+ })
+ }
+
+ override def toString: String = prettyJson
+}
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger. Each event relates to processing done for a single trigger of the streaming
+ * query. Events are emitted even when no new data is available to be processed.
+ *
+ * @param id A unique query id that persists across restarts. See `StreamingQuery.id()`.
+ * @param runId A query id that is unique for every start/restart. See `StreamingQuery.runId()`.
+ * @param name User-specified name of the query, null if not specified.
+ * @param timestamp Beginning time of the trigger in ISO8601 format, i.e. UTC timestamps.
+ * @param batchId A unique id for the current batch of data being processed. Note that in the
+ * case of retries after a failure a given batchId my be executed more than once.
+ * Similarly, when there is no data to be processed, the batchId will not be
+ * incremented.
+ * @param numInputRows The total number of records read from all the sources.
+ * @param inputRowsPerSecond The total rate at which data is arriving from all the sources.
+ * @param processedRowsPerSecond The total rate at which data from all the sources is being
+ * processed by Spark.
+ * @param batchDuration The process duration of each batch.
+ * @param durationMs The amount of time taken to perform various operations in milliseconds.
+ * @param eventTime Statistics of event time seen in this batch. It may contain the following keys:
+ * {{{
+ * "max" -> "2016-12-05T20:54:20.827Z" // maximum event time seen in this trigger
+ * "min" -> "2016-12-05T20:54:20.827Z" // minimum event time seen in this trigger
+ * "avg" -> "2016-12-05T20:54:20.827Z" // average event time seen in this trigger
+ * "watermark" -> "2016-12-05T20:54:20.827Z" // watermark used in this trigger
+ * }}}
+ * All timestamps are in ISO8601 format, i.e. UTC timestamps.
+ * @param stateOperators Information about operators in the query that store state.
+ * @param sources detailed statistics on data being read from each of the streaming sources.
+ * @since 2.1.0
+ */
+@Evolving
+class LegacyStreamingQueryProgress private[spark](
+ val id: UUID,
+ val runId: UUID,
+ val name: String,
+ val timestamp: String,
+ val batchId: Long,
+ val numInputRows: Long,
+ val inputRowsPerSecond: Double,
+ val processedRowsPerSecond: Double,
+ val batchDuration: Long,
+ val durationMs: ju.Map[String, JLong],
+ val eventTime: ju.Map[String, String],
+ val stateOperators: Array[StateOperatorProgress],
+ val sources: Array[SourceProgress],
+ val sink: SinkProgress) extends Serializable {
Review Comment:
I removed `observedMetrics` from the original class since it has some issues to deserialize the `GenericRowWithSchema` type. If we don't need this then it should be okay, otherwise I'll need to investigate this deserialization issue.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173345341
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,289 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during a trigger.
+ * @param json
+ * A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark] (
+ val operatorName: String,
+ val numRowsTotal: Long,
+ val numRowsUpdated: Long,
+ val allUpdatesTimeMs: Long,
+ val numRowsRemoved: Long,
+ val allRemovalsTimeMs: Long,
+ val commitTimeMs: Long,
+ val memoryUsedBytes: Long,
+ val numRowsDroppedByWatermark: Long,
+ val numShufflePartitions: Long,
+ val numStateStoreInstances: Long,
+ val customMetrics: ju.Map[String, JLong] = new ju.HashMap())
+ extends Serializable {
+
+ /** The compact JSON representation of this progress. */
+ def json: String = compact(render(jsonValue))
+
+ /** The pretty (i.e. indented) JSON representation of this progress. */
+ def prettyJson: String = pretty(render(jsonValue))
+
+ private[sql] def jsonValue: JValue = {
+ ("operatorName" -> JString(operatorName)) ~
+ ("numRowsTotal" -> JInt(numRowsTotal)) ~
+ ("numRowsUpdated" -> JInt(numRowsUpdated)) ~
+ ("allUpdatesTimeMs" -> JInt(allUpdatesTimeMs)) ~
+ ("numRowsRemoved" -> JInt(numRowsRemoved)) ~
+ ("allRemovalsTimeMs" -> JInt(allRemovalsTimeMs)) ~
+ ("commitTimeMs" -> JInt(commitTimeMs)) ~
+ ("memoryUsedBytes" -> JInt(memoryUsedBytes)) ~
+ ("numRowsDroppedByWatermark" -> JInt(numRowsDroppedByWatermark)) ~
+ ("numShufflePartitions" -> JInt(numShufflePartitions)) ~
+ ("numStateStoreInstances" -> JInt(numStateStoreInstances)) ~
+ ("customMetrics" -> {
+ if (!customMetrics.isEmpty) {
+ val keys = customMetrics.keySet.asScala.toSeq.sorted
+ keys.map { k => k -> JInt(customMetrics.get(k).toLong): JObject }.reduce(_ ~ _)
+ } else {
+ JNothing
+ }
+ })
+ }
+
+ override def toString: String = prettyJson
+}
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during a trigger. Each
+ * event relates to processing done for a single trigger of the streaming query. Events are
+ * emitted even when no new data is available to be processed.
+ *
+ * @param id
+ * A unique query id that persists across restarts. See `StreamingQuery.id()`.
+ * @param runId
+ * A query id that is unique for every start/restart. See `StreamingQuery.runId()`.
+ * @param name
+ * User-specified name of the query, null if not specified.
+ * @param timestamp
+ * Beginning time of the trigger in ISO8601 format, i.e. UTC timestamps.
+ * @param batchId
+ * A unique id for the current batch of data being processed. Note that in the case of retries
+ * after a failure a given batchId my be executed more than once. Similarly, when there is no
+ * data to be processed, the batchId will not be incremented.
+ * @param numInputRows
+ * The total number of records read from all the sources.
+ * @param inputRowsPerSecond
+ * The total rate at which data is arriving from all the sources.
+ * @param processedRowsPerSecond
+ * The total rate at which data from all the sources is being processed by Spark.
+ * @param batchDuration
+ * The process duration of each batch.
+ * @param durationMs
+ * The amount of time taken to perform various operations in milliseconds.
+ * @param eventTime
+ * Statistics of event time seen in this batch. It may contain the following keys:
+ * {{{
+ * "max" -> "2016-12-05T20:54:20.827Z" // maximum event time seen in this trigger
+ * "min" -> "2016-12-05T20:54:20.827Z" // minimum event time seen in this trigger
+ * "avg" -> "2016-12-05T20:54:20.827Z" // average event time seen in this trigger
+ * "watermark" -> "2016-12-05T20:54:20.827Z" // watermark used in this trigger
+ * }}}
+ * All timestamps are in ISO8601 format, i.e. UTC timestamps.
+ * @param stateOperators
+ * Information about operators in the query that store state.
+ * @param sources
+ * detailed statistics on data being read from each of the streaming sources.
+ * @since 2.1.0
+ */
+@Evolving
+class LegacyStreamingQueryProgress private[spark] (
+ val id: UUID,
+ val runId: UUID,
+ val name: String,
+ val timestamp: String,
+ val batchId: Long,
+ val numInputRows: Long,
+ val inputRowsPerSecond: Double,
+ val processedRowsPerSecond: Double,
+ val batchDuration: Long,
+ val durationMs: ju.Map[String, JLong],
+ val eventTime: ju.Map[String, String],
+ val stateOperators: Array[StateOperatorProgress],
+ val sources: Array[SourceProgress],
+ val sink: SinkProgress)
Review Comment:
I removed `observedMetrics` from the original class since it has some issues to deserialize the `GenericRowWithSchema` type. If we don't need this then it should be okay, otherwise I'll need to investigate this deserialization issue.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on PR #40895:
URL: https://github.com/apache/spark/pull/40895#issuecomment-1517370683
Conflict with https://github.com/apache/spark/pull/40892, will close this PR
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173336296
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,265 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger.
+ * @param json A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark](
Review Comment:
Mostly copied from [org.apache.spark.sql.streaming.progress.scala](https://github.com/apache/spark/blob/7b93415836057107b9e296fe79cfd67565874551/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala), did some small adjustment.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173411029
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryProgressSuite.scala:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import org.apache.spark.sql.SQLHelper
+import org.apache.spark.sql.connect.client.util.RemoteSparkSession
+
+class StreamingQueryProgressSuite extends RemoteSparkSession with SQLHelper {
+
+ test("test from json") {
Review Comment:
This case is good, I will copy to my pr, thanks @bogao007
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173344903
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,289 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during a trigger.
+ * @param json
+ * A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark] (
Review Comment:
Mostly copied from [org.apache.spark.sql.streaming.progress.scala](https://github.com/apache/spark/blob/7b93415836057107b9e296fe79cfd67565874551/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala), did some small adjustment.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173336296
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -17,6 +17,265 @@
package org.apache.spark.sql.streaming
+import java.{util => ju}
+import java.lang.{Long => JLong}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.json4s._
+import org.json4s.JsonAST.JValue
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.annotation.Evolving
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
+
+/**
+ * Information about progress made in the execution of a [[StreamingQuery]] during
+ * a trigger.
+ * @param json A json string that contains entire streaming query progress.
+ */
+@Evolving
class StreamingQueryProgress private[sql] (val json: String) {
- // TODO(SPARK-43128): (Implement full object by parsing from json).
+ private val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
+
+ /** Deserialize streaming query progress json string to [[LegacyStreamingQueryProgress]] */
+ def fromJson(): LegacyStreamingQueryProgress = {
+ jsonMapper.readValue(json, classOf[LegacyStreamingQueryProgress])
+ }
+}
+
+/**
+ * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
+ */
+@Evolving
+class StateOperatorProgress private[spark](
Review Comment:
Mostly copied from [org.apache.spark.sql.streaming.progress.scala](https://github.com/apache/spark/blob/7b93415836057107b9e296fe79cfd67565874551/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala), did some small adjustment.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 closed pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 closed pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
URL: https://github.com/apache/spark/pull/40895
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bogao007 commented on a diff in pull request #40895: [SPARK-43128] [SS] [Connect] Implemented StreamingQueryProgress for Spark Connect
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #40895:
URL: https://github.com/apache/spark/pull/40895#discussion_r1173528090
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryProgressSuite.scala:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import org.apache.spark.sql.SQLHelper
+import org.apache.spark.sql.connect.client.util.RemoteSparkSession
+
+class StreamingQueryProgressSuite extends RemoteSparkSession with SQLHelper {
+
+ test("test from json") {
+ val jsonString =
+ """{
+ | "id": "52de8a4e-1af7-4358-83e1-74d5d44bbc1b",
+ | "runId": "46680bd2-bb67-4828-90d6-3f76f43acd70",
+ | "name": "myName",
+ | "timestamp": "2023-04-20T04:52:47.890Z",
+ | "batchId": 169095,
+ | "numInputRows": 839,
+ | "inputRowsPerSecond": 0.0,
+ | "processedRowsPerSecond": 252.33082706766916,
+ | "durationMs":
+ | {
+ | "addBatch": 2240,
+ | "getBatch": 99,
+ | "latestOffset": 74,
+ | "queryPlanning": 3,
+ | "triggerExecution": 3325,
+ | "walCommit": 325
+ | },
+ | "eventTime" : {
+ | "avg" : "2016-12-05T20:54:20.827Z",
+ | "max" : "2016-12-05T20:54:20.827Z",
+ | "min" : "2016-12-05T20:54:20.827Z",
+ | "watermark" : "2016-12-05T20:54:20.827Z"
+ | },
+ | "stateOperators" : [ {
+ | "operatorName" : "op1",
+ | "numRowsTotal" : 0,
+ | "numRowsUpdated" : 1,
+ | "allUpdatesTimeMs" : 1,
+ | "numRowsRemoved" : 2,
+ | "allRemovalsTimeMs" : 34,
+ | "commitTimeMs" : 23,
+ | "memoryUsedBytes" : 3,
+ | "numRowsDroppedByWatermark" : 0,
+ | "numShufflePartitions" : 2,
+ | "numStateStoreInstances" : 2,
+ | "customMetrics" : {
+ | "loadedMapCacheHitCount" : 1,
+ | "loadedMapCacheMissCount" : 0,
+ | "stateOnCurrentVersionSizeBytes" : 2
+ | }
+ | } ],
+ | "sources" : [ {
+ | "description" : "source",
+ | "startOffset" : "123",
+ | "endOffset" : "456",
+ | "latestOffset" : "789",
+ | "numInputRows" : 678,
+ | "inputRowsPerSecond" : 10.0
+ | } ],
+ | "sink" : {
+ | "description" : "sink",
+ | "numOutputRows" : -1
+ | }
+ |}""".stripMargin
+
+ val streamingQueryProgress = new StreamingQueryProgress(jsonString)
+ val result = streamingQueryProgress.fromJson()
+ assert("52de8a4e-1af7-4358-83e1-74d5d44bbc1b" === result.id.toString)
+ assert("46680bd2-bb67-4828-90d6-3f76f43acd70" === result.runId.toString)
+ assert(839 === result.numInputRows)
Review Comment:
Yep, the data here is not consistent since I gathered from different places. You can update it if you want to add to your PR
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org