Hi,
I'm new to databricks and I'm trying to use stream for my incremental data. This data has duplicates which can be solved using a window function. Can you check where my code goes wrong?
1-------
#Using Auto Loader to read new files
schema = df1.schema
rStream = (
spark.readStream.format("parquet")
.option("includeExistingFiles", False) \
.schema(schema).load(srcpath)
)
2------
#deltatable
from delta.tables import *
deltadf = DeltaTable.forName(spark,f"hive.final_table")
3------
#Merge and Deduplication
def mergetoDF(microdf, batchId😞
print(f"inside forEachBatch for batchid:{batchId}. Rows in passed dataframe:{microdf.count()}")
microdf = Window.partitionBy("key1","key2").orderBy(col('process_key').desc())
microdf = rStream.withColumn("rownum",row_number().over(microdf)).filter("rownum = 1").drop("rownum")
(deltadf
.merge(microdf, "source.key1= target.key1and source.key2= target.key2")
.whenMatchedUpdateAll("source.process_key<> target.process_key")
.whenNotMatchedInsertAll()
.execute()
)
4------
wStream = (rStream.writeStream \
.format("delta") \
.trigger(availableNow=True) \
.option("checkpointLocation", 'path/checkpoints/') \
.option("mergeSchema","true") \
.outputMode("append") \
.foreachBatch(mergetoDF) \
.start())
I'm really confused since it's my first time using pyspark.
Looking forward to your help.
