Hi all,
I am using a Python Notebook that's being triggered by a data factory to update Delta tables each night. In short, the process is as follows:
1. A data factory orchestrates all processes. It knows which tables to process due to an array.
2. For each table, it triggers a notebook that queries the source MySQL database, and puts the data into a parquet file in the Lakehouse.
3. It then triggers another notebook for this specific file, which transfers the data from the file into a delta lake table. This is a MERGE action. So it does not create a new delta lake table, it updates an existing one.
We recently noticed that one of the source tables contained 2 new columns. We do not have explicit mapping. We just say: all columns in source should be transferred to target. I have updated the PySpark notebook with an ALTER TABLE statement, so that new columns are added in the Delta lake table, whenever they pop up in the source. You can find the full python function that I use for this below.
What happens however, when I query the table through the Lakehouse SQL endpoint, these two new columns, as well as new records of data, do not come up in the query. I can only see the records that were already in the table before we added this ALTER TABLE statement in the notebook. But the weird thing is: when I query the exact same delta lake table in a PySpark notebook, I can see all the new records, and those new columns as well.
Any idea how this is caused and how to fix it?
Thanks!
Python function:
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
def upsert_to_delta(df_source, delta_path, name_of_destination, merge_columns, is_incremental):
spark = SparkSession.builder.getOrCreate()
# Load the Delta table. If it's not a delta table yet, create it
if DeltaTable.isDeltaTable(spark, delta_path):
delta_table = DeltaTable.forPath(spark, delta_path)
df_target = delta_table.toDF()
# Check for new columns and their data type in the source
new_columns = {col: df_source.schema[col].dataType.simpleString() for col in df_source.columns if col not in df_target.columns}
else:
df_source.write.format("delta").save(delta_path)
delta_table = DeltaTable.forPath(spark, delta_path)
new_columns = {}
# Add new columns and the data type in the table, and insert NULL values
for column, dataType in new_columns.items():
spark.sql(f"ALTER TABLE delta.`{delta_path}` ADD COLUMNS ({column} {dataType})")
spark.sql(f"UPDATE delta.`{delta_path}` SET {column} = NULL")
# Define the merge operation
merge_condition = " AND ".join([f"source.{col} = target.{col}" for col in merge_columns])
# When matched and source is newer, update the target
update_mapping = {col: f"source.{col}" for col in df_source.columns}
# When not matched, insert the source row to target
insert_mapping = {col: f"source.{col}" for col in df_source.columns}
# Execute the merge operation
delta_table.alias('target') \
.merge(df_source.alias('source'), merge_condition) \
.whenMatchedUpdate(set = update_mapping) \
.whenNotMatchedInsert(values = insert_mapping) \
.execute()
# If not incremental, delete rows in target that are not present in source
if not is_incremental:
# Find the rows to delete
df_to_delete = delta_table.toDF().join(df_source, merge_columns, "left_anti")
# Delete rows
for row in df_to_delete.collect():
delete_condition = " AND ".join([f"{col} = {repr(row[col])}" for col in merge_columns])
delta_table.delete(delete_condition)
