Merge operation with ouputMode update in autoloader databricks

db_knowledge — Tue, 04 Jun 2024 14:36:49 GMT

Hi team,

I am trying to do merge operation along with outputMode('update') and foreachmode byusing below code but it is not updating data could you please any help on this?

output=(casting_df.writeStream.format('delta').trigger(availableNow=True).option(

"checkpointLocation", f"{output_path}/checkpoint"

).option("mergeSchema", "true").foreachBatch(lambda casting_df,batch_id:upsertToDelta(casting_df,batch_id)).outputMode("update").start()

)

print(f"{target_schema}.{table_name}")

def upsertToDelta(casting_df,bacthId😞

casting_df.persist()

delta_df = DeltaTable.forName(spark, f"{target_schema}.{table_name}")

(delta_df.alias("t")\

.merge(

source = casting_df.alias("m"),

condition = "m.key_column=t.key_column"

.whenMatchedUpdateAll()\

.whenNotMatchedInsertAll()\

.execute()

)

casting_df.unpersist()
Any help is appreciated

Re: Merge operation with ouputMode update in autoloader databricks

anardinelli — Tue, 04 Jun 2024 15:29:06 GMT

Hi @db_knowledge

Please try .foreachBatch(upsertToDelta) instead of creating the lambda inside it.

Best,

Alessandro

Re: Merge operation with ouputMode update in autoloader databricks

db_knowledge — Tue, 04 Jun 2024 18:33:17 GMT

Hi @anardinelli tried with foreachBatch(upsertToDelta) instead of creating the lambda inside it. but not working

topic Re: Merge operation with ouputMode update in autoloader databricks in Data Engineering

Merge operation with ouputMode update in autoloader databricks

Re: Merge operation with ouputMode update in autoloader databricks

Re: Merge operation with ouputMode update in autoloader databricks