You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xupefei (via GitHub)" <gi...@apache.org> on 2024/11/21 12:56:26 UTC
[PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
xupefei opened a new pull request, #48920:
URL: https://github.com/apache/spark/pull/48920
### What changes were proposed in this pull request?
This PR fixes an issue that some of DataFrame execution paths would not process `ObservedMetrics`. The fix is done by injecting a lazy processing logic into the result iterator.
This PR also fixes an issue in which on the Server side, two observed metrics can be assigned to the same Plan ID when they are in the same plan (e.g., one observation is used as the input of another). The fix is to traverse the plan and find all observations with correct IDs.
### Why are the changes needed?
To fix a bug.
### Does this PR introduce _any_ user-facing change?
Yes, this bug is user-facing.
### How was this patch tested?
New tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852025363
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1190,14 +1190,14 @@ class SparkConnectPlanner(
val input = transformRelation(rel.getInput)
if (input.isStreaming || executeHolderOpt.isEmpty) {
- CollectMetrics(name, metrics.map(_.named), transformRelation(rel.getInput), planId)
+ CollectMetrics(name, metrics.map(_.named), input, planId)
Review Comment:
This fixes a bug where the input of a `CollectMetrics` can be processed two times, once in Line 1190 and once here/below.
When the `input` contains another `CollectMetrics`, transforming it twice will cause two `Observation` objects (in the input) to be initialised and registered two times to the system. Since only one of them will be fulfilled when the query finishes, the one we'll be looking at may not have any data.
This issue is highlighted in the test case `Observation.get is blocked until the query is finished ...`, where we specifically execute `observedObservedDf`, which is a `CollectMetrics` that has another `CollectMetrics` as its input.
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852025363
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1190,14 +1190,14 @@ class SparkConnectPlanner(
val input = transformRelation(rel.getInput)
if (input.isStreaming || executeHolderOpt.isEmpty) {
- CollectMetrics(name, metrics.map(_.named), transformRelation(rel.getInput), planId)
+ CollectMetrics(name, metrics.map(_.named), input, planId)
Review Comment:
This fixes a bug where the input of a `CollectMetrics` can be processed two times, once in Line 1190 and once here/below.
When the `input` contains another `CollectMetrics`, transforming it twice will cause two `Observation` objects (in the input) to be initialised and registered two times to the system. Since only one of them will be fulfilled when the query finishes, the one we'll be looking at may not be the right one.
This issue is captured by the test case `Observation.get is blocked until the query is finished ...` where we only execute the `observedObservedDf`, which is a `CollectMetrics` having another `CollectMetrics` as input.
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852292409
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -409,11 +405,26 @@ class SparkSession private[sql] (
val plan = proto.Plan.newBuilder().setCommand(command).build()
// .toSeq forces that the iterator is consumed and closed. On top, ignore all
// progress messages.
- client.execute(plan).filter(!_.hasExecutionProgress).toSeq
+ execute(plan).filter(!_.hasExecutionProgress).toSeq
}
- private[sql] def execute(plan: proto.Plan): CloseableIterator[ExecutePlanResponse] =
- client.execute(plan)
+ /**
+ * The real `execute` method that calls into `SparkConnectClient`.
+ *
+ * Here we inject a lazy map to process registered observed metrics, so consumers of the returned
+ * iterator does not need to worry about it.
+ *
+ * Please make sure all `execute` overrides call this method.
+ */
+ private[sql] def execute(plan: proto.Plan): CloseableIterator[ExecutePlanResponse] = {
Review Comment:
Would it make sense to rename this to `executeInternal` so it is clear that all the overrides are calling this specific method?
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852025363
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1190,14 +1190,14 @@ class SparkConnectPlanner(
val input = transformRelation(rel.getInput)
if (input.isStreaming || executeHolderOpt.isEmpty) {
- CollectMetrics(name, metrics.map(_.named), transformRelation(rel.getInput), planId)
+ CollectMetrics(name, metrics.map(_.named), input, planId)
Review Comment:
This fixes a bug where the input of a `CollectMetrics` can be processed two times, once in Line 1190 and once here/below.
When the `input` contains another `CollectMetrics`, transforming it twice will cause two `Observation` objects (in the input) to be initialised and registered two times to the system. Since only one of them will be fulfilled when the query finishes, the one we'll be looking at may not be the right one.
--
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
Re: [PR] [SPARK-50372][CONNECT][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48920:
URL: https://github.com/apache/spark/pull/48920#issuecomment-2513181352
Merged to master.
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852025363
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1190,14 +1190,14 @@ class SparkConnectPlanner(
val input = transformRelation(rel.getInput)
if (input.isStreaming || executeHolderOpt.isEmpty) {
- CollectMetrics(name, metrics.map(_.named), transformRelation(rel.getInput), planId)
+ CollectMetrics(name, metrics.map(_.named), input, planId)
Review Comment:
This fixes a bug where the input of a `CollectMetrics` can be processed two times, once in Line 1190 and once here/below.
When the `input` contains another `CollectMetrics`, transforming it twice will cause two `Observation` objects (in the input) to be initialised and registered two times to the system. Since only one of them will be fulfilled when the query finishes, the one we'll be looking at may not have any data.
This issue is captured by the test case `Observation.get is blocked until the query is finished ...` where we only execute the `observedObservedDf`, which is a `CollectMetrics` having another `CollectMetrics` as input.
--
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
Re: [PR] [SPARK-50372][Connect][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852456594
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -409,11 +405,26 @@ class SparkSession private[sql] (
val plan = proto.Plan.newBuilder().setCommand(command).build()
// .toSeq forces that the iterator is consumed and closed. On top, ignore all
// progress messages.
- client.execute(plan).filter(!_.hasExecutionProgress).toSeq
+ execute(plan).filter(!_.hasExecutionProgress).toSeq
}
- private[sql] def execute(plan: proto.Plan): CloseableIterator[ExecutePlanResponse] =
- client.execute(plan)
+ /**
+ * The real `execute` method that calls into `SparkConnectClient`.
+ *
+ * Here we inject a lazy map to process registered observed metrics, so consumers of the returned
+ * iterator does not need to worry about it.
+ *
+ * Please make sure all `execute` overrides call this method.
+ */
+ private[sql] def execute(plan: proto.Plan): CloseableIterator[ExecutePlanResponse] = {
Review Comment:
Good idea. Done!
--
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
Re: [PR] [SPARK-50372][CONNECT][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #48920: [SPARK-50372][CONNECT][SQL] Make all DF execution path collect observed metrics
URL: https://github.com/apache/spark/pull/48920
--
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
Re: [PR] [SPARK-50372][CONNECT][SQL] Make all DF execution path collect observed metrics [spark]
Posted by "xupefei (via GitHub)" <gi...@apache.org>.
xupefei commented on PR #48920:
URL: https://github.com/apache/spark/pull/48920#issuecomment-2494652947
> Looks good but mind listing up affected API in the PR description?
Done!
--
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