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!
