You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/11/18 21:08:55 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request, #38719: [SPARK-41999][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
HeartSaVioR opened a new pull request, #38719:
URL: https://github.com/apache/spark/pull/38719
### What changes were proposed in this pull request?
This PR proposes to fix the metrics issue for streaming query when DSv1 streaming source and DSv2 streaming source are co-used. If the streaming query has both DSv1 streaming source and DSv2 streaming source, only DSv1 streaming source produced correct metrics.
There is a bug in ProgressReporter that it tries to match logical node for DSv2 streaming source with OffsetHolder, which will be never matched. Given that physical node for DSv2 streaming source contains both source information and metrics, we can simply deduce all the necessary information from the physical node rather than trying to find the source from association map.
### Why are the changes needed?
The logic of collecting metrics does not collect metrics for DSv2 streaming sources properly.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New test case.
--
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] HeartSaVioR commented on a diff in pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38719:
URL: https://github.com/apache/spark/pull/38719#discussion_r1026994040
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala:
##########
@@ -345,7 +345,14 @@ trait ProgressReporter extends Logging {
val allExecPlanLeaves = lastExecution.executedPlan.collectLeaves()
if (allLogicalPlanLeaves.size == allExecPlanLeaves.size) {
val execLeafToSource = allLogicalPlanLeaves.zip(allExecPlanLeaves).flatMap {
- case (lp, ep) => logicalPlanLeafToSource.get(lp).map { source => ep -> source }
+ case (_, ep: MicroBatchScanExec) =>
+ // SPARK-41199: `logicalPlanLeafToSource` contains OffsetHolder instance for DSv2
+ // streaming source, hence we cannot lookup the actual source from the map.
+ // The physical node for DSv2 streaming source contains the information of the source
+ // by itself, so leverage it.
+ Some(ep -> ep.stream)
Review Comment:
So all the problems from streaming query metrics are due to DSv1 which we do not have a dedicated logical node / physical node for source.
--
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] HeartSaVioR commented on pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38719:
URL: https://github.com/apache/spark/pull/38719#issuecomment-1320531883
cc. @zsxwing @viirya @xuanyuanking Please take a look. Thanks in advance!
--
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] viirya commented on a diff in pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #38719:
URL: https://github.com/apache/spark/pull/38719#discussion_r1026993007
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala:
##########
@@ -345,7 +345,14 @@ trait ProgressReporter extends Logging {
val allExecPlanLeaves = lastExecution.executedPlan.collectLeaves()
if (allLogicalPlanLeaves.size == allExecPlanLeaves.size) {
val execLeafToSource = allLogicalPlanLeaves.zip(allExecPlanLeaves).flatMap {
- case (lp, ep) => logicalPlanLeafToSource.get(lp).map { source => ep -> source }
+ case (_, ep: MicroBatchScanExec) =>
+ // SPARK-41199: `logicalPlanLeafToSource` contains OffsetHolder instance for DSv2
+ // streaming source, hence we cannot lookup the actual source from the map.
+ // The physical node for DSv2 streaming source contains the information of the source
+ // by itself, so leverage it.
+ Some(ep -> ep.stream)
Review Comment:
Why it is only for DS v1 mixing with DS v2? Seems DS v2 stream source always cannot be matched by `logicalPlanLeafToSource` because it is `OffsetHolder`.
--
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] HeartSaVioR commented on pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38719:
URL: https://github.com/apache/spark/pull/38719#issuecomment-1320888432
Thanks! Merging 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
[GitHub] [spark] viirya commented on a diff in pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #38719:
URL: https://github.com/apache/spark/pull/38719#discussion_r1026999005
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala:
##########
@@ -345,7 +345,14 @@ trait ProgressReporter extends Logging {
val allExecPlanLeaves = lastExecution.executedPlan.collectLeaves()
if (allLogicalPlanLeaves.size == allExecPlanLeaves.size) {
val execLeafToSource = allLogicalPlanLeaves.zip(allExecPlanLeaves).flatMap {
- case (lp, ep) => logicalPlanLeafToSource.get(lp).map { source => ep -> source }
+ case (_, ep: MicroBatchScanExec) =>
+ // SPARK-41199: `logicalPlanLeafToSource` contains OffsetHolder instance for DSv2
+ // streaming source, hence we cannot lookup the actual source from the map.
+ // The physical node for DSv2 streaming source contains the information of the source
+ // by itself, so leverage it.
+ Some(ep -> ep.stream)
Review Comment:
Oh, got it. I saw that there is a `onlyDataSourceV2Sources` and a dedicated block for it.
--
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] HeartSaVioR commented on a diff in pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38719:
URL: https://github.com/apache/spark/pull/38719#discussion_r1026993747
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala:
##########
@@ -345,7 +345,14 @@ trait ProgressReporter extends Logging {
val allExecPlanLeaves = lastExecution.executedPlan.collectLeaves()
if (allLogicalPlanLeaves.size == allExecPlanLeaves.size) {
val execLeafToSource = allLogicalPlanLeaves.zip(allExecPlanLeaves).flatMap {
- case (lp, ep) => logicalPlanLeafToSource.get(lp).map { source => ep -> source }
+ case (_, ep: MicroBatchScanExec) =>
+ // SPARK-41199: `logicalPlanLeafToSource` contains OffsetHolder instance for DSv2
+ // streaming source, hence we cannot lookup the actual source from the map.
+ // The physical node for DSv2 streaming source contains the information of the source
+ // by itself, so leverage it.
+ Some(ep -> ep.stream)
Review Comment:
We have two different paths - if there are only DSv2 streaming sources, we don't even try to match the logical plan and physical plan. We just collect the metrics out from physical plan, which is always accurate.
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala#L296-L316
--
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] HeartSaVioR closed pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #38719: [SPARK-41199][SS] Fix metrics issue when DSv1 streaming source and DSv2 streaming source are co-used
URL: https://github.com/apache/spark/pull/38719
--
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