Skip to content

[SPARK-40460][SS] Fix streaming metrics when selecting _metadata #37905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Yaohua628 wants to merge 4 commits into from
Closed

[SPARK-40460][SS] Fix streaming metrics when selecting _metadata #37905

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
995ba39
fix metrics
Yaohua628 Sep 16, 2022
046adc7
comments
Yaohua628 Sep 16, 2022
881f867
self union and join
Yaohua628 Sep 17, 2022
f19d7c9
oops
Yaohua628 Sep 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.streaming

import scala.collection.mutable.{Map => MutableMap}
import scala.collection.mutable

import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.catalyst.encoders.RowEncoder
Expand Down Expand Up @@ -553,7 +554,7 @@ class MicroBatchExecution(
logDebug(s"Running batch $currentBatchId")

// Request unprocessed data from all sources.
newData = reportTimeTaken("getBatch") {
val mutableNewData = mutable.Map() ++ reportTimeTaken("getBatch") {
Comment thread
Yaohua628 marked this conversation as resolved.
Outdated
availableOffsets.flatMap {
case (source: Source, available: Offset)
if committedOffsets.get(source).map(_ != available).getOrElse(true) =>
Expand Down Expand Up @@ -590,7 +591,7 @@ class MicroBatchExecution(
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
mutableNewData.get(source).map { dataPlan =>

@HeartSaVioR HeartSaVioR Sep 16, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While we are here, probably less intrusive change would be moving (L594 ~ L610) to L567. After the change we wouldn't need to make a change to newData here.

@Yaohua628 Yaohua628 Sep 16, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, I initially thought about that, but we need to know the output from StreamingExecutionRelation(source, output, catalogTable) to resolve _metadata right (L591 ~ L593)?

@HeartSaVioR HeartSaVioR Sep 16, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, you're right. I missed that.

Btw, looks like my change (tagging catalogTable into LogicalRelation) will also fall into this bug. Thanks for fixing this.

@Yaohua628 Yaohua628 Sep 16, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Np - an unintentional fix :-)
Thanks for helping!

@HeartSaVioR HeartSaVioR Sep 16, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe, we may want to check the case of self-union / self-join to verify we really didn't break things. This works only when this condition is true leaf : source = 1 : 1 (otherwise we are overwriting the value in map), while the code comment of ProgressReporter tells there are counter cases.

@Yaohua628 Yaohua628 Sep 16, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it - could you share an example? In this case, does that mean the leaf : source = 1 : N?

@HeartSaVioR HeartSaVioR Sep 17, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The code comment actually doesn't say much and I'm speculating. Let's just try a best effort, self-union and self-join. df = spark.readStream... -> df.union(df) / df = spark.readStream... -> df.join(df)

val hasFileMetadata = output.exists {

@cloud-fan cloud-fan Sep 16, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looking at the code, seems the problem is we resolve the metadata columns in every micro-batch. Shouldn't we only resolve it once?

@HeartSaVioR HeartSaVioR Sep 16, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It will require Source to indicate the request of metadata column and produce the logical plan accordingly when getBatch is called. My understanding is that DSv1 source does not have an interface to receive the information of which columns will be referred in actual query.

case FileSourceMetadataAttribute(_) => true
case _ => false
Expand All @@ -608,6 +609,11 @@ class MicroBatchExecution(
}
newRelation
}
// SPARK-40460: overwrite the entry with the new logicalPlan
// because it might contain the _metadata column. It is a necessary change,
// in the ProgressReporter, we use the following mapping to get correct streaming metrics:
// streaming logical plan (with sources) <==> trigger's logical plan <==> executed plan
mutableNewData.put(source, finalDataPlan)
val maxFields = SQLConf.get.maxToStringFields
assert(output.size == finalDataPlan.output.size,
s"Invalid batch: ${truncatedString(output, ",", maxFields)} != " +
Expand All @@ -623,14 +629,14 @@ class MicroBatchExecution(

// For v2 sources.
case r: StreamingDataSourceV2Relation =>
newData.get(r.stream).map {
mutableNewData.get(r.stream).map {
case OffsetHolder(start, end) =>
r.copy(startOffset = Some(start), endOffset = Some(end))
}.getOrElse {
LocalRelation(r.output, isStreaming = true)
}
}

newData = mutableNewData.toMap
// Rewire the plan to use the new attributes that were returned by the source.
val newAttributePlan = newBatchesPlan.transformAllExpressionsWithPruning(
_.containsPattern(CURRENT_LIKE)) {
Expand Down
6 changes: 4 additions & 2 deletions ...e/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import org.apache.spark.sql.{AnalysisException, Column, DataFrame, QueryTest, Ro
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}

Expand Down Expand Up @@ -524,10 +525,11 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
.select("*", "_metadata")
.writeStream.format("json")
.option("checkpointLocation", dir.getCanonicalPath + "/target/checkpoint")
.trigger(Trigger.Once())
Comment thread
Yaohua628 marked this conversation as resolved.
Outdated
.start(dir.getCanonicalPath + "/target/new-streaming-data")

stream.processAllAvailable()
stream.stop()
stream.awaitTermination()
assert(stream.lastProgress.numInputRows == 2L)

val newDF = spark.read.format("json")
.load(dir.getCanonicalPath + "/target/new-streaming-data")
Expand Down