Haripriya created SPARK-26154:
---------------------------------
Summary: Stream-stream joins - left outer join gives inconistent
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
Environment: Spark version - Spark 2.3.2
OS- Suse 11
Reporter: Haripriya
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 "abc" is processed. But again in batch 5, null value
is provided for same data
Steps:
In spark-shell
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
Step2 : Start producing data
bin/kafka-console-producer.sh --broker-list ip:9092 --topic topic1
>abc
>def
>ghi
>123
>123
>234
>345
/kafka-console-producer.sh --broker-list ip:9092 --topic topic2
>abc
>ghi
>123
>345
>234
>678
Output obtained:
-------------------------------------------
Batch: 0
-------------------------------------------
+----+----------+-----+-----------+
|data|recordTime|data1|recordTime1|
+----+----------+-----+-----------+
+----+----------+-----+-----------+
-------------------------------------------
Batch: 1
-------------------------------------------
+----+----------+-----+-----------+
|data|recordTime|data1|recordTime1|
+----+----------+-----+-----------+
+----+----------+-----+-----------+
-------------------------------------------
Batch: 2
-------------------------------------------
+----+-----------------------+-----+-----------------------+
|data|recordTime |data1|recordTime1 |
+----+-----------------------+-----+-----------------------+
|ghi |2018-11-02 10:52:26.072|ghi |2018-11-02 10:52:28.309|
|abc |2018-11-02 10:52:18.627|abc |2018-11-02 10:52:22.249|
+----+-----------------------+-----+-----------------------+
Batch: 3
-------------------------------------------
+----+-----------------------+-----+-----------------------+
|data|recordTime |data1|recordTime1 |
+----+-----------------------+-----+-----------------------+
|123 |2018-11-02 10:52:31.062|123 |2018-11-02 10:52:33.094|
+----+-----------------------+-----+-----------------------+
-------------------------------------------
Batch: 4
-------------------------------------------
+----+-----------------------+-----+-----------------------+
|data|recordTime |data1|recordTime1 |
+----+-----------------------+-----+-----------------------+
|345 |2018-11-02 10:52:41.252|345 |2018-11-02 10:52:44.178|
+----+-----------------------+-----+-----------------------+
-------------------------------------------
Batch: 5
-------------------------------------------
+----+-----------------------+-----+-----------------------+
|data|recordTime |data1|recordTime1 |
+----+-----------------------+-----+-----------------------+
|678 |2018-11-02 10:53:04.116|678 |2018-11-02 10:53:06.275|
|abc |2018-11-02 10:52:18.627|null |null |
|def |2018-11-02 10:52:24.296|null |null |
+----+-----------------------+-----+-----------------------+
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
