Zaimplementuj wolno zmieniający się wymiar (SCD) typ 2

Po wybraniu typu SCD 2 dla wymiaru należy zaimplementować strukturę tabeli i logikę przechwytywania zmian w usłudze Azure Databricks. Ta jednostka koncentruje się na tworzeniu tabel typu 2 SCD i implementowaniu wzorców MERGE, które utrzymują historię wersji wraz ze zmianami danych źródłowych.

Tworzenie tabeli SCD Type 2

Poniższy kod SQL tworzy tabelę wymiarów klienta z kolumnami śledzenia SCD Type 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
);

Klauzula GENERATED ALWAYS AS IDENTITY tworzy automatycznie zwiększany klucz zastępczy dla każdego nowego wiersza. Włączenie strumienia danych o zmianach umożliwia wydajne przechwytywanie zmian przyrostowych przez procesy zależne.

Implementowanie przechwytywania zmian za pomocą funkcji MERGE

Instrukcja MERGE zapewnia wydajny sposób implementowania logiki TYPU 2 protokołu SCD podczas przetwarzania aktualizacji z systemów źródłowych. Instrukcja obsługuje wstawianie, aktualizacje i logikę przechowywania wersji w jednej transakcji.

Następujący wzorzec zamyka bieżącą wersję i wstawia nową wersję po wystąpieniu zmian:

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;

Wykonywanie zapytań dotyczących danych historycznych

Po zaimplementowaniu tabeli SCD Type 2 można wykonywać zapytania dotyczące danych, które istniały w dowolnym momencie. Wzorzec zapytania zależy od potrzeb analitycznych.

Zapytania dotyczące konkretnego momentu w czasie

Aby znaleźć stan danych w określonym momencie, przefiltruj okres ważności:

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';

To zapytanie zwraca dokładnie jeden wiersz na klienta — wersję, która była ważna w określonym momencie czasowym.

Historia rekordów śledzenia

Aby wyświetlić pełną historię zmian dla określonej jednostki:

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

To zapytanie ujawnia wszystkie wersje klienta C-555, pokazujące, jak ich atrybuty zmieniły się wraz z upływem czasu.

Użyj podróży w czasie Delta Lake

Usługa Delta Lake oferuje wbudowane możliwości podróży w czasie, które uzupełniają jawne projekty tabel typu SCD Type 2. Możesz wykonywać zapytania dotyczące poprzednich wersji tabeli przy użyciu składni TIMESTAMP AS OF lub 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;