Merge operation with ouputMode update in autoloader databricks

db_knowledge
New Contributor II

‎06-04-2024 07:35 AM - edited ‎06-04-2024 07:36 AM

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
0 Kudos

anardinelli
Databricks Employee
Databricks Employee

‎06-04-2024 08:29 AM

Hi @db_knowledge 

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

Best,

Alessandro

0 Kudos

db_knowledge
New Contributor II
In response to anardinelli

‎06-04-2024 11:33 AM

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

0 Kudos