Implementare una dimensione a modifica lenta (SCD) di tipo 2

Completato

Dopo aver selezionato SCD Type 2 per una dimensione, è necessario implementare la struttura della tabella e modificare la logica di acquisizione in Azure Databricks. Questa unità è incentrata sulla creazione di tabelle scD type 2 e sull'implementazione dei modelli MERGE che mantengono la cronologia delle versioni man mano che cambiano i dati di origine.

Creare la tabella SCD Type 2

Il seguente codice SQL crea una tabella dimensionale del cliente con colonne di tracciamento di tipo SCD 2.

CREATE TABLE sales.customer (
    customer_sk BIGINT GENERATED ALWAYS AS IDENTITY,
    customer_id STRING NOT NULL,
    customer_name STRING,
    email STRING,
    city STRING,
    region STRING,
    valid_from TIMESTAMP NOT NULL,
    valid_to TIMESTAMP NOT NULL,
    is_current BOOLEAN
)
USING DELTA
TBLPROPERTIES (
    delta.enableChangeDataFeed = true
);

La GENERATED ALWAYS AS IDENTITY clausola crea una chiave surrogata a incremento automatico per ogni nuova riga. L'abilitazione del feed di dati delle modifiche consente ai processi downstream di acquisire in modo efficiente le modifiche incrementali.

Implementare l'acquisizione delle modifiche con MERGE

L'istruzione MERGE fornisce un modo efficiente per implementare la logica scD Type 2 durante l'elaborazione degli aggiornamenti dai sistemi di origine. L'istruzione gestisce inserimenti, aggiornamenti e logica di controllo delle versioni in una singola transazione.

Il modello seguente chiude la versione corrente e inserisce la nuova versione in caso di modifiche:

MERGE INTO sales.customer AS target
USING (
    SELECT 
        source.customer_id,
        source.customer_name,
        source.email,
        source.city,
        source.region,
        current_timestamp() AS valid_from,
        CAST('9999-12-31' AS TIMESTAMP) AS valid_to,
        true AS is_current
    FROM staging.customers AS source
) AS updates
ON target.customer_id = updates.customer_id AND target.is_current = true
WHEN MATCHED AND (
    target.customer_name <> updates.customer_name OR
    target.email <> updates.email OR
    target.city <> updates.city OR
    target.region <> updates.region
) THEN UPDATE SET
    target.valid_to = current_timestamp(),
    target.is_current = false
WHEN NOT MATCHED THEN INSERT (
    customer_id, customer_name, email, city, region,
    valid_from, valid_to, is_current
) VALUES (
    updates.customer_id, updates.customer_name, updates.email, 
    updates.city, updates.region, updates.valid_from, 
    updates.valid_to, updates.is_current
);

-- Insert new versions for updated records
INSERT INTO sales.customer
SELECT 
    s.customer_id,
    s.customer_name,
    s.email,
    s.city,
    s.region,
    current_timestamp() AS valid_from,
    CAST('9999-12-31' AS TIMESTAMP) AS valid_to,
    true AS is_current
FROM staging.customers s
JOIN sales.customer h 
    ON s.customer_id = h.customer_id 
    AND h.valid_to = current_timestamp()
    AND h.is_current = false;

Suggerimento

Prendere in considerazione l'uso di pipeline Lakeflow con l'API AUTO CDC per l'elaborazione automatizzata SCD di tipo 2. Questo approccio gestisce i record non ordinati e semplifica la manutenzione della tabella scD type 2. Consultare la documentazione di Azure Databricks per le pipeline di Change Data Capture.

Eseguire query sui dati cronologici

Dopo aver implementato una tabella SCD Type 2, è possibile eseguire query sui dati così come esistevano in qualsiasi momento. Il modello di query dipende dalle esigenze analitiche.

Query temporizzato

Per trovare lo stato dei dati in un momento specifico, filtrare in base al periodo di validità:

SELECT customer_id, customer_name, city
FROM sales.customer
WHERE valid_from <= '2023-06-15 12:00:00'
  AND valid_to > '2023-06-15 12:00:00';

Questa query restituisce esattamente una riga per cliente, ovvero la versione valida al timestamp specificato.

Storico dei risultati

Per visualizzare la cronologia completa delle modifiche per un'entità specifica:

SELECT customer_name, city, valid_from, valid_to
FROM sales.customer
WHERE customer_id = 'C-555'
ORDER BY valid_from;

Questa query rivela tutte le versioni di C-555, mostrando come i loro attributi sono cambiati nel tempo.

Usare Delta Lake con viaggio nel tempo

Delta Lake offre funzionalità predefinite di viaggio nel tempo che completano i progetti espliciti di tabelle SCD Type 2. È possibile eseguire query sulle versioni precedenti della tabella usando la TIMESTAMP AS OF sintassi o VERSION AS OF :

-- Query table state from 7 days ago
SELECT * FROM sales.customer 
TIMESTAMP AS OF '2024-01-15';

-- Query a specific table version
SELECT * FROM sales.customer 
VERSION AS OF 42;

Importante

Il viaggio nel tempo di Delta Lake ha una retention predefinita di 7 giorni. Per un'analisi cronologica più lunga, usare colonne SCD esplicite (ValidFrom, ValidTo) anziché basarsi esclusivamente sul viaggio in tempo. Configurare delta.logRetentionDuration e delta.deletedFileRetentionDuration se è necessario l'accesso a tempo prolungato.