[
https://issues.apache.org/jira/browse/SPARK-27433?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16816105#comment-16816105
]
Binit commented on SPARK-27433:
-------------------------------
This is a blocker for us as the basic functionality of the left_outer join is
not working. Do you guys have any date in mind when you are trying to provide a
patch for this fix ?
> Spark Structured Streaming left outer joins returns outer nulls for already
> matched rows
> ----------------------------------------------------------------------------------------
>
> Key: SPARK-27433
> URL: https://issues.apache.org/jira/browse/SPARK-27433
> Project: Spark
> Issue Type: Question
> Components: Structured Streaming
> Affects Versions: 2.3.0
> Reporter: Binit
> Priority: Blocker
>
> I m basically using the example given in Spark's the documentation here:
> [https://spark.apache.org/docs/2.3.0/structured-streaming-programming-guide.html#outer-joins-with-watermarking]
> with the built-in test stream in which one stream is ahead by 3 seconds (was
> originally using kafka but ran into the same issue). The results returned the
> match columns correctly, however after a while the same key is returned with
> an outer null.
> Is this the expected behavior? Is there a way to exclude the duplicate outer
> null results when there was a match?
> Code:
> {{val testStream = session.readStream.format("rate") .option("rowsPerSecond",
> "5").option("numPartitions", "1").load() val impressions = testStream
> .select( (col("value") + 15).as("impressionAdId"),
> col("timestamp").as("impressionTime")) val clicks = testStream .select(
> col("value").as("clickAdId"), col("timestamp").as("clickTime")) // Apply
> watermarks on event-time columns val impressionsWithWatermark =
> impressions.withWatermark("impressionTime", "20 seconds") val
> clicksWithWatermark = clicks.withWatermark("clickTime", "30 seconds") // Join
> with event-time constraints val result = impressionsWithWatermark.join(
> clicksWithWatermark, expr(""" clickAdId = impressionAdId AND clickTime >=
> impressionTime AND clickTime <= impressionTime + interval 10 seconds """),
> joinType = "leftOuter" // can be "inner", "leftOuter", "rightOuter" ) val
> query =
> result.writeStream.outputMode("update").format("console").option("truncate",
> false).start() query.awaitTermination()}}
> Result:
> {{------------------------------------------- Batch: 19
> -------------------------------------------
> +--------------+-----------------------+---------+-----------------------+
> |impressionAdId|impressionTime |clickAdId|clickTime |
> +--------------+-----------------------+---------+-----------------------+
> |100 |2018-05-23 22:18:38.362|100 |2018-05-23 22:18:41.362| |101 |2018-05-23
> 22:18:38.562|101 |2018-05-23 22:18:41.562| |102 |2018-05-23 22:18:38.762|102
> |2018-05-23 22:18:41.762| |103 |2018-05-23 22:18:38.962|103 |2018-05-23
> 22:18:41.962| |104 |2018-05-23 22:18:39.162|104 |2018-05-23 22:18:42.162|
> +--------------+-----------------------+---------+-----------------------+
> ------------------------------------------- Batch: 57
> -------------------------------------------
> +--------------+-----------------------+---------+-----------------------+
> |impressionAdId|impressionTime |clickAdId|clickTime |
> +--------------+-----------------------+---------+-----------------------+
> |290 |2018-05-23 22:19:16.362|290 |2018-05-23 22:19:19.362| |291 |2018-05-23
> 22:19:16.562|291 |2018-05-23 22:19:19.562| |292 |2018-05-23 22:19:16.762|292
> |2018-05-23 22:19:19.762| |293 |2018-05-23 22:19:16.962|293 |2018-05-23
> 22:19:19.962| |294 |2018-05-23 22:19:17.162|294 |2018-05-23 22:19:20.162|
> |100 |2018-05-23 22:18:38.362|null |null | |99 |2018-05-23 22:18:38.162|null
> |null | |103 |2018-05-23 22:18:38.962|null |null | |101 |2018-05-23
> 22:18:38.562|null |null | |102 |2018-05-23 22:18:38.762|null |null |
> +--------------+-----------------------+---------+-----------------------+}}
> {{This question is also asked in the stackoverflow. Please find the link
> below}}
> {{[https://stackoverflow.com/questions/50500111/spark-structured-streaming-left-outer-joins-returns-outer-nulls-for-already-matc/55616902#55616902]}}
> {{ }}
> {{101 & 103 have already come in the join but still it is coming in the outer
> left join.}}
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
