Spark Streaming Issues while performing left join

RamanP9404 — Wed, 13 Dec 2023 13:08:36 GMT

Hi team,

I'm struck in a Spark Structured streaming use-case.

Requirement: To read two streaming data frames, perform a left join on it and display the results.

Issue: While performing a left join, the resultant data frame contains only rows where there was a match and discards the rest of unmatched rows.

Ex:

Left table - 200 rows, Right table - 150 rows

Final output - 150 rows(These are the ones with key matches).

Code Snippet:

As stream-stream join requires defining watermark column and event-time range conditions, I have applied the same in the below code.

My data sources used in this example are static tables(table 1, table 2), but just to process just the incremental data I used spark.readStream() while performing the read. Kindly note there are no duplicates on either tables.

# COMMAND ----------

bronze_df1 = spark.readStream.table("catalog.schema.table1").withColumn("load_timestamp", current_timestamp())
display(bronze_df1) #200 rows
# COMMAND ----------

bronze_df2 = spark.readStream.table("catalog.schema.table2").withColumn("load_timestamp", current_timestamp())
display(bronze_df2) #150 rows
# COMMAND ----------

bronze_df1 = bronze_df1.withWatermark("load_timestamp","5 minutes")
bronze_df1.createOrReplaceTempView("bronze_df1_view") # We can create a view out of streaming dataframe as per Spark Streaming documentation
display(spark.sql("select * from bronze_df1_view")) #200 rows
# COMMAND ----------

bronze_df2 = bronze_df2.withWatermark("load_timestamp","5 minutes")
bronze_df2.createOrReplaceTempView("bronze_df2_view")
display(spark.sql("select * from bronze_df2_view"))

# COMMAND ----------

# Define the SQL query to transform the
final_df = spark.sql(f"""

select
a.id,
a.asset_type,
a.country_code,
a.state_code,
b.id, --This can be null,
a.load_timestamp
FROM bronze_df1_view a
LEFT OUTER JOIN bronze_df2_view b
ON

a.id = b.id and
b.load_timestamp between a.load_timestamp - interval 10 minutes and a.load_timestamp + interval 10 minutes

--Giving a bigger event-time range just to ensure none of the data gets ruled out

""")

display(final_df) #As soon as the execution starts, I could see 150 matched rows and rest of the unmatched rows never get printed on the console despite letting the job run for a longer time.

P.S: As mentioned earlier, mine is a very simple streaming use-case (Just to process incremental data). Since the source tables are static tables, I don't expect any data arriving late or out-of-order. I have tried using multiple values in watermark column ranging from 0 seconds to 1 hours, yet every time I see left join results are same as inner join.

P.P.S: I have been stuck into this issue for over a month now, any heads up you can provide will be much appreciated. Thanks!

topic Spark Streaming Issues while performing left join in Get Started Discussions

Spark Streaming Issues while performing left join