You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Siying Dong (Jira)" <ji...@apache.org> on 2024/04/30 20:50:00 UTC
[jira] [Commented] (SPARK-48023) Stream-stream join with windowed+watermarked dropDuplicates suppresses all rows
[ https://issues.apache.org/jira/browse/SPARK-48023?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17842501#comment-17842501 ]
Siying Dong commented on SPARK-48023:
-------------------------------------
I'm not able to run `.withWatermark("window"....)`. It throws an exception? Is there a typo? I tried to replace "window" with "ts" but then the result doesn't repro.
If stream-stream join results have been written to `path`. How is stream-stream join related?
If it is not related, can you provide a shorter repro?
> Stream-stream join with windowed+watermarked dropDuplicates suppresses all rows
> -------------------------------------------------------------------------------
>
> Key: SPARK-48023
> URL: https://issues.apache.org/jira/browse/SPARK-48023
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 3.5.0
> Reporter: Scott Schenkein
> Priority: Major
>
> When applying a streaming dropDuplicates-with-watermark to a self-referential stream-stream left-join with watermark, the dropDuplicates drops all rows. If the watermark is eliminated from the dropDuplicates, the query behaves as expected.
>
> The code below demonstrates the error:
>
> {code:java}
> #
> # 1. Generate the test data
> #
> size = 1000
> step_secs = 300
> event_offset_secs = 240
> # Add some lines to not left join
> skips = set()
> skips.add(3)
> skips.add(18)
> skips.add(800)
> def lit_ts(secs):
> return datetime.datetime.fromtimestamp(secs)
> data = []
> base_time = time.time()
> for x in range(size):
> ts = base_time + (step_secs * x)
> data.append({"event_id": f"one_{x}", "join_id": x, "ts": lit_ts(ts)})
> if x not in skips:
> data.append(
> {
> "event_id": f"two_{x}",
> "join_id": x,
> "ts": lit_ts(ts - event_offset_secs),
> }
> )
> # Add duplicates to validate the dropDuplicates
> for i in range(len(data)):
> data.append(data[i].copy())
> #
> # 2. Write the results so we can stream
> #
> path = "/tmp/bugtest"
> df = spark.createDataFrame(data, schema="event_id string, join_id string, ts TIMESTAMP")
> df.repartition(1).write.format("delta").mode("overwrite").save(path)
> df = spark.read.format("delta").load(path)
> df.createOrReplaceTempView("test_data")
> #
> # 3. Define the test query
> #
> sql = """
> with also_test_data as (
> select * from test_data
> where event_id like 'two%'
> )
> select td.* from test_data
> td left join also_test_data on (
> td.join_id = also_test_data.join_id
> and also_test_data.ts >= td.ts - INTERVAL 15 MINUTES
> and also_test_data.ts <= td.ts
> )
> where td.event_id like 'one%'
> -- rows where left-join does not match
> and also_test_data.event_id is NULL
> """
> #
> # 4. Run it non-streaming w/non-deDuplicated to validate results
> #
> res = spark.sql(sql)
> print("Static query")
> res.show(truncate=False)
> # Static query
> # +--------+-------+-------------------------+
> # |event_id|join_id|ts |
> # +--------+-------+-------------------------+
> # |one_3 |3 |2024-04-27 17:36:54.97927|
> # |one_18 |18 |2024-04-27 18:51:54.97927|
> # |one_800 |800 |2024-04-30 12:01:54.97927|
> # |one_3 |3 |2024-04-27 17:36:54.97927|
> # |one_18 |18 |2024-04-27 18:51:54.97927|
> # |one_800 |800 |2024-04-30 12:01:54.97927|
> # +--------+-------+-------------------------+
> #
> # 5. Run it as a stream with no-dropDuplicates
> #
> def write_stream(res):
> (
> res.writeStream.outputMode("append")
> .trigger(availableNow=True)
> .format("console")
> .start()
> .awaitTermination()
> )
> sdf = spark.readStream.format("delta").load(path).withWatermark('ts', '15 minutes')
> sdf.createOrReplaceTempView("test_data")
> res = spark.sql(sql)
> write_stream(res)
> # Batch: 1
> # -------------------------------------------
> # +--------+-------+--------------------+
> # |event_id|join_id| ts|
> # +--------+-------+--------------------+
> # | one_800| 800|2024-04-30 12:01:...|
> # | one_800| 800|2024-04-30 12:01:...|
> # | one_3| 3|2024-04-27 17:36:...|
> # | one_3| 3|2024-04-27 17:36:...|
> # | one_18| 18|2024-04-27 18:51:...|
> # | one_18| 18|2024-04-27 18:51:...|
> # +--------+-------+--------------------+
> #
> # 6. Run it as a stream with dropDuplicates, but no extra watermark
> #
> sdf = spark.readStream.format("delta").load(path).withWatermark("ts", "15 minutes")
> sdf.createOrReplaceTempView("test_data")
> res = spark.sql(sql)
> res = (
> (res.select(window("ts", "30 minutes"), "*"))
> .dropDuplicates(["event_id"]).drop("window")
> )
> write_stream(res)
> # -------------------------------------------
> # Batch: 1
> # -------------------------------------------
> # +--------+-------+--------------------+
> # |event_id|join_id| ts|
> # +--------+-------+--------------------+
> # | one_800| 800|2024-04-30 12:01:...|
> # | one_18| 18|2024-04-27 18:51:...|
> # | one_3| 3|2024-04-27 17:36:...|
> # +--------+-------+--------------------+
> #
> # 7. Run it as a stream with dropDuplicates, using a watermark
> # THIS is where we see the error
> #
> sdf = spark.readStream.format("delta").load(path).withWatermark("ts", "15 minutes")
> sdf.createOrReplaceTempView("test_data")
> res = spark.sql(sql)
> res = (
> (res.select(window("ts", "30 minutes"), "*"))
> .withWatermark("window", '30 minutes')
> .dropDuplicates(["event_id"]).drop("window")
> )
> write_stream(res)
> print("END")
> # -------------------------------------------
> # Batch: 0
> # -------------------------------------------
> # +--------+-------+---+
> # |event_id|join_id| ts|
> # +--------+-------+---+
> # +--------+-------+---+
> # END{code}
>
>
--
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