You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Richard Swinbank (Jira)" <ji...@apache.org> on 2024/07/11 08:57:00 UTC
[jira] [Updated] (SPARK-48869) Pivoting watermarked streaming dataframe fails with correctness error
[ https://issues.apache.org/jira/browse/SPARK-48869?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Richard Swinbank updated SPARK-48869:
-------------------------------------
Description:
When I run this query:
{code:python}
from pyspark.sql.functions import col, first
df_src = (
spark.readStream
.format("rate")
.option("rowsPerSecond", 3)
.load()
.withColumn("text_value", col("value").cast("string"))
.withColumn("label", col("value"))
)
df_pivot = (
df_src
.withWatermark("timestamp", "10 minutes")
.groupBy("label")
.pivot("value", ["x", "y"])
.agg(first("value"))
)
display(df_pivot)
{code}
it fails with error {_}AnalysisException: Detected pattern of possible 'correctness' issue due to global watermark{_}.
This happens because the analyzer [resolves the pivot into two aggregate operations|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L817]. Multiple aggregations in a watermarked streaming pipeline [cause this error to be thrown|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala#L123] to prevent correctness issues caused by stateful operations using a single global watermark.
I've raised this as a bug because of the pivot failure, but please advise if that's inappropriate! Some observations:
* We could say that pivot is not supported in a watermarked streaming pipeline. That would be disappointing, and it's not always true (depending on the datatype of the column being pivoted - for example, pivoting a string column causes the pivot to be implemented as a one-step aggregate).
* I don't understand the internals well enough to know whether the two-step aggregate is vulnerable to global watermark correctness issues at all. If not, it would seem to me that the unsupported operation check is too aggressive - it would need to be able to distinguish between the general multiple stateful operation problem and the adjacent aggregates implemented by the analyzer for pivot.
* If the two-step aggregate for pivot is safe, one can disable the correctness check with {{{}spark.conf.set("spark.sql.streaming.statefulOperator.checkCorrectness.enabled", "false"){}}}, but that affects all operations - in principle a query could contain pivot operations *and* multiple operations vulnerable to global watermark correctness.
Many thanks
was:
When I run this query:
{code:python}
from pyspark.sql.functions import col, first
df_src = (
spark.readStream
.format("rate")
.option("rowsPerSecond", 3)
.load()
.withColumn("text_value", col("value").cast("string"))
.withColumn("label", col("text_value"))
)
df_pivot = (
df_src
.withWatermark("timestamp", "10 minutes")
.groupBy("label")
.pivot("value", ["x", "y"])
.agg(first("value"))
)
display(df_pivot)
{code}
it fails with error {_}AnalysisException: Detected pattern of possible 'correctness' issue due to global watermark{_}.
This happens because the analyzer [resolves the pivot into two aggregate operations|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L817]. Multiple aggregations in a watermarked streaming pipeline [cause this error to be thrown|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala#L123] to prevent correctness issues caused by stateful operations using a single global watermark.
I've raised this as a bug because of the pivot failure, but please advise if that's inappropriate! Some observations:
* We could say that pivot is not supported in a watermarked streaming pipeline. That would be disappointing, and it's not always true (depending on the datatype of the column being pivoted - for example, pivoting a string column causes the pivot to be implemented as a one-step aggregate).
* I don't understand the internals well enough to know whether the two-step aggregate is vulnerable to global watermark correctness issues at all. If not, it would seem to me that the unsupported operation check is too aggressive - it would need to be able to distinguish between the general multiple stateful operation problem and the adjacent aggregates implemented by the analyzer for pivot.
* If the two-step aggregate for pivot is safe, one can disable the correctness check with {{{}spark.conf.set("spark.sql.streaming.statefulOperator.checkCorrectness.enabled", "false"){}}}, but that affects all operations - in principle a query could contain pivot operations *and* multiple operations vulnerable to global watermark correctness.
Many thanks
> Pivoting watermarked streaming dataframe fails with correctness error
> ---------------------------------------------------------------------
>
> Key: SPARK-48869
> URL: https://issues.apache.org/jira/browse/SPARK-48869
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 3.5.0
> Environment: Azure Databricks runtime 14.3 LTS
> Reporter: Richard Swinbank
> Priority: Minor
>
> When I run this query:
> {code:python}
> from pyspark.sql.functions import col, first
> df_src = (
> spark.readStream
> .format("rate")
> .option("rowsPerSecond", 3)
> .load()
> .withColumn("text_value", col("value").cast("string"))
> .withColumn("label", col("value"))
> )
> df_pivot = (
> df_src
> .withWatermark("timestamp", "10 minutes")
> .groupBy("label")
> .pivot("value", ["x", "y"])
> .agg(first("value"))
> )
> display(df_pivot)
> {code}
>
> it fails with error {_}AnalysisException: Detected pattern of possible 'correctness' issue due to global watermark{_}.
> This happens because the analyzer [resolves the pivot into two aggregate operations|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L817]. Multiple aggregations in a watermarked streaming pipeline [cause this error to be thrown|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala#L123] to prevent correctness issues caused by stateful operations using a single global watermark.
> I've raised this as a bug because of the pivot failure, but please advise if that's inappropriate! Some observations:
> * We could say that pivot is not supported in a watermarked streaming pipeline. That would be disappointing, and it's not always true (depending on the datatype of the column being pivoted - for example, pivoting a string column causes the pivot to be implemented as a one-step aggregate).
> * I don't understand the internals well enough to know whether the two-step aggregate is vulnerable to global watermark correctness issues at all. If not, it would seem to me that the unsupported operation check is too aggressive - it would need to be able to distinguish between the general multiple stateful operation problem and the adjacent aggregates implemented by the analyzer for pivot.
> * If the two-step aggregate for pivot is safe, one can disable the correctness check with {{{}spark.conf.set("spark.sql.streaming.statefulOperator.checkCorrectness.enabled", "false"){}}}, but that affects all operations - in principle a query could contain pivot operations *and* multiple operations vulnerable to global watermark correctness.
> Many thanks
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org