Performing upsert in notebook PySpark (python)

Hi, I have created a similar function for this. Hope it can help you with some modifications. 

Function Overview: The upsert_table function updates or inserts data into a target table based on the given DataFrame (df_new), load type and if table exists or not.
Full Load: If specified or if the table doesn't exist, it overwrites the existing table or creates a new one with the new data.
Incremental Load: If the table exists and a full reload is not specified, it merges the new data into the existing table. This involves:

• Updating existing records that match between the new data and the existing table.
• Inserting new records that don't find a match in the existing table.

Parameters:
df_new: your DF with new data that you want to upsert into target table, cleaned with same columns as target tables (the ones you specify).
load_type: A string indicating the type of load operation. It can either be "FULL" for a complete overwrite of the existing table or not specified for an incremental update where new rows are merged with existing data.
I have this as I do full loads periodically, so I control this parameter from pipelines. If you don't have such requirements you can modify the code to get rid of it. 
target_table_path_loc: The file path location where the target table is stored or should be created. This path is used to construct the full path to the target table. For me its only Tables/ as its in my default lakehouse 
target_table_name: The name of the target table. This name is standardized to lowercase in the function to ensure consistency in how table names are handled.

def upsert_table(df_new, load_type, target_table_path_loc, target_table_name):
    target_table_name = target_table_name.lower()
    target_table_path = target_table_path_loc + target_table_name
    load_type = load_type.upper()
    numOfNewRows = df_new.count() #I use this for some validation / logging etc.
    try:
        if not dt.DeltaTable.isDeltaTable(spark, target_table_path) or load_type == "FULL": 
            df_new.write.format("delta").mode("overwrite").saveAsTable(target_table_name)
        else:
            current_dt = DeltaTable.forPath(spark, target_table_path)
            current_dt.alias("target").merge(
            df_new.alias("source"),
            "target.SearchId = source.SearchId")\
            .whenMatchedUpdateAll()\
            .whenNotMatchedInsertAll()\
            .execute()
    except Exception as e: 
			....handle your exceptions....