Trying to incrementally upsert changes into a Delta table us

Question

Trying to incrementally upsert changes into a Delta table using autoloader and foreachBatch as described here <https docs databricks com structured streaming delta lake html upsert from streaming queries using foreachbatch> It seems my updated records get appended and not updated whatever I try Should the inital Delta table be set with certain properties When I use outputMode Update in the writeStream I get Data source com databricks sql transaction tahoe sources DeltaDataSource does not support Update output mode

Roel Knitel · Answer

`def upsert data microBatchOutputDF batchId ` deltaTable = DeltaTable forPath spark lt some abfss path silver gt deltaTable alias data merge df alias newData newData entryNo = data entryNo whenMatchedUpdateAll whenNotMatchedInstertAll execute def autoload datasource source format table name datatarget query = spark readStream format cloudFiles option cloudFiles format source format option inferSchema True option cloudFiles schemaLocation datasource+ checkpoint load datasource writeStream trigger once=True foreachBatch upsert data format delta outputMode update queryName data incrementeeel laden van brons naar zilver option checkpointLocation datasource+ checkpoint start datatarget return query autoload lt some abfss path bronze gt csv Entries lt some abfss path silver gt

Dominique Brezinski · Answer

Using foreachBatch you can ignore outputMode foreachBatch decouples the dataframe from the stream output In the code you posted microBatchOutputDF is not used Is there missing code where it is transformed renamed into df

Dominique Brezinski · Answer

The behavior you describe would happen if your merge predicate is never true

Roel Knitel · Answer

Thanks for pointing that out Dominiqe Totally missed that Got it working now slightly smiling face thanks again