[
https://issues.apache.org/jira/browse/SPARK-26154?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17021627#comment-17021627
]
Jungtaek Lim commented on SPARK-26154:
--------------------------------------
Leaving information why this issue cannot be easily ported back to 2.x version
line although it fixes the correctness.
We fixed the issue with "destructive way" on existing query as the state in
existing query cannot be corrected. In 3.0 migration guide we added below
content:
{quote}Spark 3.0 fixes the correctness issue on Stream-stream outer join, which
changes the schema of state. (SPARK-26154 for more details) Spark 3.0 will fail
the query if you start your query from checkpoint constructed from Spark 2.x
which uses stream-stream outer join. Please discard the checkpoint and replay
previous inputs to recalculate outputs.
{quote}
End users might not be mad if "major" version requires them to lose something.
They still have Spark 2.x version line to deny the change and take the risk (as
the issue is occurred from "edge-case"). If we put this to 2.x version line
they may have no way to deny.
Please note that unlike other states having versions which can be co-existed,
for stream-stream outer join, there're only "valid" (state format version = 2)
and "invalid" (state format version = 1) state format which cannot be
co-existed.
We cannot still simply drop state format version 1, because stream-stream inner
join is not affected by this bug and we don't want to let the case also discard
the state. It will affect too many existing queries and I'd like to reduce the
impact.
> Stream-stream joins - left outer join gives inconsistent output
> ---------------------------------------------------------------
>
> Key: SPARK-26154
> URL: https://issues.apache.org/jira/browse/SPARK-26154
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 2.3.2, 3.0.0
> Environment: Spark version - Spark 2.3.2
> OS- Suse 11
> Reporter: Haripriya
> Assignee: Jungtaek Lim
> Priority: Blocker
> Labels: correctness
> Fix For: 3.0.0
>
>
> Stream-stream joins using left outer join gives inconsistent output
> The data processed once, is being processed again and gives null value. In
> Batch 2, the input data "3" is processed. But again in batch 6, null value
> is provided for same data
> Steps
> In spark-shell
> {code:java}
> scala> import org.apache.spark.sql.functions.{col, expr}
> import org.apache.spark.sql.functions.{col, expr}
> scala> import org.apache.spark.sql.streaming.Trigger
> import org.apache.spark.sql.streaming.Trigger
> scala> val lines_stream1 = spark.readStream.
> | format("kafka").
> | option("kafka.bootstrap.servers", "ip:9092").
> | option("subscribe", "topic1").
> | option("includeTimestamp", true).
> | load().
> | selectExpr("CAST (value AS String)","CAST(timestamp AS
> TIMESTAMP)").as[(String,Timestamp)].
> | select(col("value") as("data"),col("timestamp")
> as("recordTime")).
> | select("data","recordTime").
> | withWatermark("recordTime", "5 seconds ")
> lines_stream1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] =
> [data: string, recordTime: timestamp]
> scala> val lines_stream2 = spark.readStream.
> | format("kafka").
> | option("kafka.bootstrap.servers", "ip:9092").
> | option("subscribe", "topic2").
> | option("includeTimestamp", value = true).
> | load().
> | selectExpr("CAST (value AS String)","CAST(timestamp AS
> TIMESTAMP)").as[(String,Timestamp)].
> | select(col("value") as("data1"),col("timestamp")
> as("recordTime1")).
> | select("data1","recordTime1").
> | withWatermark("recordTime1", "10 seconds ")
> lines_stream2: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] =
> [data1: string, recordTime1: timestamp]
> scala> val query = lines_stream1.join(lines_stream2, expr (
> | """
> | | data == data1 and
> | | recordTime1 >= recordTime and
> | | recordTime1 <= recordTime + interval 5 seconds
> | """.stripMargin),"left").
> | writeStream.
> | option("truncate","false").
> | outputMode("append").
> | format("console").option("checkpointLocation",
> "/tmp/leftouter/").
> | trigger(Trigger.ProcessingTime ("5 seconds")).
> | start()
> query: org.apache.spark.sql.streaming.StreamingQuery =
> org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@1a48f55b
> {code}
> Step2 : Start producing data
> kafka-console-producer.sh --broker-list ip:9092 --topic topic1
> >1
> >2
> >3
> >4
> >5
> >aa
> >bb
> >cc
> kafka-console-producer.sh --broker-list ip:9092 --topic topic2
> >2
> >2
> >3
> >4
> >5
> >aa
> >cc
> >ee
> >ee
>
> Output obtained:
> {code:java}
> Batch: 0
> -------------------------------------------
> +----+----------+-----+-----------+
> |data|recordTime|data1|recordTime1|
> +----+----------+-----+-----------+
> +----+----------+-----+-----------+
> -------------------------------------------
> Batch: 1
> -------------------------------------------
> +----+----------+-----+-----------+
> |data|recordTime|data1|recordTime1|
> +----+----------+-----+-----------+
> +----+----------+-----+-----------+
> -------------------------------------------
> Batch: 2
> -------------------------------------------
> +----+-----------------------+-----+-----------------------+
> |data|recordTime |data1|recordTime1 |
> +----+-----------------------+-----+-----------------------+
> |3 |2018-11-22 20:09:35.053|3 |2018-11-22 20:09:36.506|
> |2 |2018-11-22 20:09:31.613|2 |2018-11-22 20:09:33.116|
> +----+-----------------------+-----+-----------------------+
> -------------------------------------------
> Batch: 3
> -------------------------------------------
> +----+-----------------------+-----+-----------------------+
> |data|recordTime |data1|recordTime1 |
> +----+-----------------------+-----+-----------------------+
> |4 |2018-11-22 20:09:38.654|4 |2018-11-22 20:09:39.818|
> +----+-----------------------+-----+-----------------------+
> -------------------------------------------
> Batch: 4
> -------------------------------------------
> +----+-----------------------+-----+-----------------------+
> |data|recordTime |data1|recordTime1 |
> +----+-----------------------+-----+-----------------------+
> |5 |2018-11-22 20:09:44.809|5 |2018-11-22 20:09:47.452|
> |1 |2018-11-22 20:09:22.662|null |null |
> +----+-----------------------+-----+-----------------------+
> -------------------------------------------
> Batch: 5
> -------------------------------------------
> +----+-----------------------+-----+-----------------------+
> |data|recordTime |data1|recordTime1 |
> +----+-----------------------+-----+-----------------------+
> |cc |2018-11-22 20:10:06.654|cc |2018-11-22 20:10:08.701|
> |aa |2018-11-22 20:10:01.536|aa |2018-11-22 20:10:03.259|
> +----+-----------------------+-----+-----------------------+
> -------------------------------------------
> Batch: 6
> -------------------------------------------
> +----+-----------------------+-----+-----------+
> |data|recordTime |data1|recordTime1|
> +----+-----------------------+-----+-----------+
> |3 |2018-11-22 20:09:35.053|null |null |
> +----+-----------------------+-----+-----------+
> {code}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
