Yaoxuan Wu created SPARK-60784:
----------------------------------
Summary: Structured Streaming: dropDuplicates / distinct after an
aggregation in Complete output mode drops groups from later micro-batches
Key: SPARK-60784
URL: https://issues.apache.org/jira/browse/SPARK-60784
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 4.1.1, 4.2.0, 3.5.8, 4.0.0
Reporter: Yaoxuan Wu
{code:python}
import tempfile
d = tempfile.mkdtemp()
spark.createDataFrame([(1,), (2,)], "k
int").coalesce(1).write.mode("append").parquet(d) # file 1
spark.createDataFrame([(2,), (3,)], "k
int").coalesce(1).write.mode("append").parquet(d) # file 2
batches = {}
q = (spark.readStream.schema("k int").option("maxFilesPerTrigger", 1).parquet(d)
.groupBy("k").count().dropDuplicates()
.writeStream.outputMode("complete")
.foreachBatch(lambda df, i: batches.__setitem__(i, sorted(df.collect())))
.trigger(availableNow=True).start())
q.awaitTermination()
print(batches[max(batches)])
# expected (the complete result table, = the batch query): [Row(k=1, count=1),
Row(k=2, count=2), Row(k=3, count=1)]
# actual: [Row(k=2, count=2),
Row(k=3, count=1)]
{code}
In Complete mode every micro-batch should contain the whole result table. The
aggregation does emit all groups in each batch, but the following
{{dropDuplicates}} (or {{distinct()}}) keeps its keys in state across batches
and drops every group row it saw before unchanged, so k = 1 disappears from the
second batch.
Cause: {{UnsupportedOperationChecker.checkForStreaming}} only requires an
aggregation for Complete mode ({{case InternalOutputModes.Complete if
aggregates.isEmpty}}); a {{Deduplicate}} after the aggregation is not checked
({{containsCompleteData}} is only used for {{Sort}}), and
{{isStatefulOperation}} counts a {{Deduplicate}} as stateful only when its keys
have an event-time column, so the chained-stateful-operator check does not see
it either. At runtime {{StateStoreSaveExec}} in Complete mode returns all
groups each batch and {{StreamingDeduplicateExec}} filters them by
{{store.keyExists(key)}}. The query should either be rejected or the
deduplication should not keep state across batches in Complete mode.
Reproduces on 3.5.8, 4.0.0, 4.1.1 and 4.2.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
