[Python] SCD2 Duplicates issue pyspark - notebooks

I tried to run the last cell of my notebooks after basic transformation. However, it insert and generate duplicates whenever it runs instead of adding or updates value. I tried to keep it simple for a single table.

def process_scd2(spark, bronze_df, silver_path, business_cols, silver_table_path):
"""
Processes SCD2 logic using SQL

Args:
bronze_df: The input DataFrame from the Bronze layer.
silver_path: The path to the Silver layer.
business_cols: A list of business columns to track for changes.
silver_table_path: The name of the Silver Delta table.
"""
today = date.today()

# Create a temporary view of the Bronze data for SQL queries
bronze_df.createOrReplaceTempView("bronze_data")

# 1. Create incoming data with SCD2 columns
incoming_df = spark.sql(f"""
SELECT
*,
CAST(current_timestamp() AS TIMESTAMP) AS snapshot_date,
CAST(current_timestamp() AS TIMESTAMP) AS valid_from,
CAST(NULL AS TIMESTAMP) AS valid_to,
CAST(TRUE AS BOOLEAN) AS is_current
FROM bronze_data
""")
incoming_df.createOrReplaceTempView("incoming_data")

silver_exists = False
try:
silver_df = spark.read.format("delta").load(silver_path)
silver_df.createOrReplaceTempView("silver_data")
silver_exists = True
print("Silver table loaded and temporary view created.")

except Exception:
print(f"Silver table does not exist or could not be loaded: {e}")
pass

if silver_exists:
# Identify changes
change_detection_condition = " OR ".join(
[f"(s.{col} IS NULL AND i.{col} IS NOT NULL) OR (s.{col} IS NOT NULL AND i.{col} IS NULL) OR (s.{col} != i.{col})"
for col in business_cols if col != "bookableresourceid"]
)
print(f"Change detection condition: {change_detection_condition}")


changed_bookableresourceid_df = spark.sql(f"""
SELECT DISTINCT i.bookableresourceid
FROM silver_data s
JOIN incoming_data i ON s.bookableresourceid = i.bookableresourceid
WHERE {change_detection_condition}
""")
changed_bookableresourceid_df.createOrReplaceTempView("changed_ids")
print("Changed bookableresourceid identified and temporary view created.")
changed_bookableresourceid_df.show()

# Close old versions
updates_df = spark.sql(f"""
SELECT
s.*,
CAST('{today}' AS DATE) AS valid_to,
CAST(FALSE AS BOOLEAN) AS is_current
FROM silver_data s
JOIN changed_ids c ON s.bookableresourceid = c.bookableresourceid
WHERE s.is_current = TRUE
""")
print("Old versions updated and temporary view created.")
updates_df.show()

# Insert new/updated versions
new_records_df = spark.sql(f"""
SELECT
i.*
FROM incoming_data i
LEFT JOIN silver_data s ON i.bookableresourceid = s.bookableresourceid AND s.is_current = TRUE
WHERE s.bookableresourceid IS NULL
OR i.bookableresourceid IN (SELECT bookableresourceid FROM changed_ids)
""")

print("New/updated versions selected.")
new_records_df.show()


# Combine old (updated) and new records
final_df = updates_df.unionByName(new_records_df, allowMissingColumns=True)
print("Old and new records combined.")
final_df.show()
else:
final_df = incoming_df


final_df = final_df.selectExpr(list(dict.fromkeys(final_df.columns)))

#Save table in Lakehouse
try:
final_df.write \
.format("delta") \
.mode("overwrite") \
.option("overwriteSchema", "true") \
.saveAsTable(silver_table_path)
print(f"Table '{silver_table_path}' written successfully.")
except AnalysisException as e:
print(f"Failed to write table '{silver_table_path}': {e}")
except Exception as e:
print(f"Unexpected error writing table '{silver_table_path}': {e}")


I am doing in it wrong and how can I see output ? is there a simplier version doing with pyspark SQL

Continue reading...