Scd

SCD Type-2 (Slowly Changing Dimension) merge logic for Delta Lake tables.

Requires the target table to have been created with track_history=True, meaning it contains the five SCD columns: HashSleutel, HashWaarde, GeldigVan, GeldigTot, IsActief.

delta-spark is provided by Fabric and Databricks automatically. For local testing, install it manually alongside PySpark.

scamander.transformations.scd.apply_scd2_merge(spark: SparkSession, source_df: DataFrame, target_table: str, key_columns: list[str], tracked_columns: list[str] | None = None) → dict[str, int]

Perform a full SCD Type-2 merge against an existing Delta table.

Algorithm:

  1. Compute HashSleutel (from key_columns) and HashWaarde (from tracked_columns) on the source DataFrame.

  2. Expire active target rows whose HashWaarde has changed (set IsActief=false, GeldigTot=now).

  3. Insert source rows that have no currently active target row — covers both brand-new entities and updated versions of records closed in step 2.

Entities in the target absent from source are left untouched (no hard-delete semantics). Handle explicit deletes separately if needed.

Parameters:

  • spark – Active SparkSession.

  • source_df – Incoming data. Must contain all key_columns and any columns listed in tracked_columns.

  • target_table – Fully qualified Delta table name (e.g. ‘silver.customer’).

  • key_columns – Columns that uniquely identify an entity.

  • tracked_columns – Columns whose changes trigger a new SCD row. Defaults to all non-key columns in source_df.

Returns:

Dict with ‘closed’ (rows expired in step 2) and ‘inserted’ (rows added in step 3) counts.

Veroorzaakt:

  • ImportError – When delta-spark is not available in the environment.

  • ValueError – When key_columns is empty.

Example:

from scamander.transformations.scd import apply_scd2_merge

stats = apply_scd2_merge(
    spark=platform.spark,
    source_df=silver_df,
    target_table="gold.customer",
    key_columns=["customer_id"],
    tracked_columns=["name", "address", "status"],
)
# stats -> {"closed": 3, "inserted": 5}
scamander.transformations.scd.compute_hash(df: DataFrame, columns: list[str], alias: str) → DataFrame

Add a SHA-256 hash column built from a pipe-delimited concat of the named columns.

Nulls are cast to the literal string ‘null’ so the hash is always non-null and comparisons are safe.

Parameters:

  • df – Input DataFrame.

  • columns – Columns to include in the hash.

  • alias – Name for the resulting hash column.

Returns:

DataFrame with the new hash column appended.

Example:

compute_hash(df, columns=["customer_id", "email"], alias="HashSleutel")