Structured Streaming Stream Stream Join

Question

I have logic as below using Spark Structured Streaming 2.3: Where I join two streams on id and then output the join stream data. Not using watermarking.This is a simple socket stream setup.

  val df1 = spark.readStream.format("socket").option("host","localhost").option("port","5431").load()

  val df2 = df1.as[String].map(x=>x.split(","))

  val df3 = df2.select($"value"(0).as("name"),$"value"(1).cast(IntegerType).as("id"),$"value"(2).cast(TimestampType).as("ts"))

  val df1_1 = spark.readStream.format("socket").option("host","localhost").option("port","5430").load()

  val df2_1 = df1_1.as[String].map(x=>x.split(","))

  val df3_1 = df2_1.select($"value"(0).as("name"),$"value"(1).cast(IntegerType).as("id"),$"value"(2).cast(TimestampType).as("ts"))

  val joindf = df3.join(df3_1,df3("id") <=> df3_1("id"))

  val res = joindf.writeStream.outputMode("append").trigger(Trigger.ProcessingTime(15 seconds))
    .format("console").option("truncate","false").start()

  res.awaitTermination()

Say the first trigger has data in two streams as :

df3:

vinyas,1,2018-03-17 09:04:21
namratha,2,2018-03-17 09:04:23
varsha,3,2018-03-17 09:04:33

df3_1:

vinyas,1,2018-03-17 09:04:21
shetty,2,2018-03-17 09:04:23
varsha,3,2018-03-17 09:04:33

I got the result as expected :

-------------------------------------------
Batch: 0
-------------------------------------------
+--------+---+-------------------+------+---+-------------------+
|name    |id |ts                 |name  |id |ts                 |
+--------+---+-------------------+------+---+-------------------+
|vinyas  |1  |2018-03-17 09:04:21|vinyas|1  |2018-03-17 09:04:21|
|varsha  |3  |2018-03-17 09:04:33|varsha|3  |2018-03-17 09:04:33|
|namratha|2  |2018-03-17 09:04:23|shetty|2  |2018-03-17 09:04:23|
+--------+---+-------------------+------+---+-------------------+

Next i stream data : df3:

shrinivas,2,2018-03-17 09:04:23

df3_1:

vidya,2,2018-03-17 09:04:23

Get output as :

-------------------------------------------
Batch: 1
-------------------------------------------
+---------+---+-------------------+------+---+-------------------+
|name     |id |ts                 |name  |id |ts                 |
+---------+---+-------------------+------+---+-------------------+
|shrinivas|2  |2018-03-17 09:04:23|shetty|2  |2018-03-17 09:04:23|
|namratha |2  |2018-03-17 09:04:23|vidya |2  |2018-03-17 09:04:23|
|shrinivas|2  |2018-03-17 09:04:23|vidya |2  |2018-03-17 09:04:23|
+---------+---+-------------------+------+---+-------------------+

Can someone explain how i am getting the Batch 1 result,should it not have 4 records instead of 3?

Mikhail Dubkov Mikhail Dubkov · Accepted Answer · 2018-08-11T00:07:35

According to official Spark Structured Streaming Programming Guide:

Inner Joins with optional Watermarking

Inner joins on any kind of columns along with any kind of join conditions are supported. However, as the stream runs, the size of streaming state will keep growing indefinitely as all past input must be saved as any new input can match with any input from the past. To avoid unbounded state, you have to define additional join conditions such that indefinitely old inputs cannot match with future inputs and therefore can be cleared from the state. In other words, you will have to do the following additional steps in the join.

Define watermark delays on both inputs such that the engine knows how delayed the input can be (similar to streaming aggregations)

Define a constraint on event-time across the two inputs such that the engine can figure out when old rows of one input is not going to be required (i.e. will not satisfy the time constraint) for matches with the other input. This constraint can be defined in one of the two ways.

Time range join conditions (e.g. ...JOIN ON leftTime BETWEN rightTime AND rightTime + INTERVAL 1 HOUR),

Join on event-time windows (e.g. ...JOIN ON leftTimeWindow = rightTimeWindow).

From your example, you've got correct result In Append mode:

You already have this record in previous batch result:

|namratha|2 |2018-03-17 09:04:23|shetty|2 |2018-03-17 09:04:23|

My understanding is that it should not appear anymore in Append Mode.

Hope it helps!

Structured Streaming Stream Stream Join

1 Answers