Issue with Upsert Table Action in Metadata-Driven Pipeline to Lakehouse Table

Hello parasol54 - 

Here is an example of how you can perform upserts on lakehouse tables within a metadata-driven pipline.

• Run a notebook which contains this code below. 

Create variables which define the paths (parameters for each component are passed to the notebook from the data pipeline):

# build absolute paths to lakehouse storage locations
source__path = f"{source__protocol}://{source__storage_account}@{source__endpoint}/{source__container}/{source__directory}/"
destination__path = f"{destination__protocol}://{destination__storage_account}@{destination__endpoint}/{destination__container}/{destination__directory}/"

Define a function to perform the upsert.

def process_table(source__lakehouse_path, source__table_name, destination__lakehouse_path, destination__table_name, unique_id):
    # load the tables
    source__table = DeltaTable.forPath(source__lakehouse_path+source__table_name)
    destination__table = DeltaTable.forPath(destination__lakehouse_path+destination__table_name)
    # convert the source table to a dataframe
    df = source__table.toDF()
    # define a list of columns names
    column_names = df.columns
    # perform the merge (upsert) operation
    destination__table.alias('destination') \
    .merge(
        df.alias('updates'),
        'destination.{0} = updates.{0}'.format(unique_id)
    )\
    .whenMatchedUpdate(set = {col: f"updates.{col}" for col in column_names}) \
    .whenNotMatchedInsert(values = {col: f"updates.{col}" for col in column_names})\
    .execute()

Invoke the function for each row in the metadata/control/watermark table (json_array)

for i in json_array:
    source__table_name = i["source__table"]
    destination__table_name = i["destination__table"]
    unique_id = i["unique_id"]
    process_table(source__path, source__table_name, destination__path, destination__table_name, unique_id)