ETL w usłudze Databricks SQL

W przypadku obsługi dużych ilości danych potrzebna jest ścieżka przetwarzania, która przetwarza się jedynie nowe i zmienione rekordy danych, zamiast ponownie przetwarzać cały zestaw danych. Jest to nazywane przyrostowym ETL. W usłudze Databricks SQL można tworzyć przyrostowe potoki ETL przy użyciu tabel strumieniowych i zmaterializowanych widoków bez konieczności pisania kodu proceduralnego ani ręcznego planowania odświeżeń.

Ten samouczek przeprowadzi Cię przez typowy wzorzec: śledzenie zmian produktów w czasie. Tworzysz tabelę źródłową, rejestrujesz zdarzenia zmian, budujesz tabelę wymiarową, która zachowuje pełną historię każdego produktu, utrzymujesz tabelę faktów, która stopniowo nadpisuje ostatnie rekordy, i dodajesz warstwę raportowania agregatu na górze. Używasz AUTO CDC dla tabeli wymiarów oraz przepływu REPLACE WHERE dla tabeli faktów.

Kluczową funkcją w tym samouczku jest AUTO CDC. W tradycyjnym magazynie danych można napisać złożone zapytania MERGE INTO do uzgadniania zdarzeń wstawiania, aktualizowania i usuwania w tabeli docelowej. Takie podejście jest podatne na błędy, zwłaszcza gdy zdarzenia docierają w niewłaściwej kolejności. AUTO CDC obsługuje to za Ciebie. Deklarujesz klucz biznesowy, kolumnę sekwencjonowania oraz określasz, czy chcesz typ SCD 1 (tylko najnowsza wartość) czy SCD 2 (pełna historia), a Azure Databricks automatycznie stosuje odpowiednią logikę scalania. Aby zapoznać się z przeglądem usługi CDC, zapoznaj się z AUTO CDC API: Upraszczanie przechwytywania zmian danych za pomocą potoków.

Po ukończeniu tego samouczka będziesz mieć następujące elementy:

  1. Utworzono tabelę źródłową, która śledzi zmiany za pomocą zestawienia zmian danych.
  2. Sprawdzono surowe dane dotyczące zmiany, aby zrozumieć strumień zdarzeń CDC.
  3. Użyto AUTO CDC do utworzenia tabeli wymiaru SCD typu 2 z tych zdarzeń.
  4. Zdarzenia usuwania przetwarzane przyrostowo w przepływie danych.
  5. Utworzono tabelę faktów, która stopniowo nadpisuje ostatnie wiersze za pomocą przepływu REPLACE WHERE.
  6. Utworzono zmaterializowany widok, który przyrostowo utrzymuje raport agregujący.
  7. Skonfigurowano SCHEDULE REFRESH EVERY 1 DAY, aby zmiany propagowały się automatycznie w potoku.

Wymagania

Aby ukończyć ten samouczek, musisz spełnić następujące wymagania:

Krok 1. Konfigurowanie katalogu i schematu

Otwórz edytor Databricks SQL i ustaw swój domyślny katalog oraz schemat. Musisz mieć uprawnienia do USE katalogu i wybranego schematu:

USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;

Krok 2. Tworzenie tabeli źródłowej i ładowanie danych

Utwórz tabelę products z włączoną funkcją Use change data feed on Azure Databricks (CDF). CDF to funkcja usługi Delta Lake, która rejestruje każde wstawianie, aktualizowanie i usuwanie jako dziennik zmian z możliwością wykonywania zapytań. Jest to podobne do strumienia CDC z transakcyjnego systemu źródłowego, z wyjątkiem tego, że zmiany są przechwytywane bezpośrednio w tabeli Delta, a nie z dziennika zewnętrznego. Używasz tutaj CDF, aby generować zdarzenia zmiany, które będą wykorzystywane przez dalszy potok.

  1. Utwórz tabelę i załaduj początkowe rekordy:

    CREATE OR REPLACE TABLE products (
  product_id INT,
  product_name STRING,
  category STRING,
  warehouse STRING
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

INSERT INTO products VALUES
  (1, 'Spoon', 'Cutlery', 'Seattle'),
  (2, 'Fork', 'Cutlery', 'Portland'),
  (3, 'Knife', 'Cutlery', 'Denver'),
  (4, 'Chair', 'Furniture', 'Austin'),
  (5, 'Table', 'Furniture', 'Chicago'),
  (6, 'Lamp', 'Lighting', 'Boston'),
  (7, 'Mug', 'Kitchenware', 'Seattle'),
  (8, 'Plate', 'Kitchenware', 'Atlanta'),
  (9, 'Bowl', 'Kitchenware', 'Dallas'),
  (10, 'Glass', 'Kitchenware', 'Phoenix');

  2. Zasymuluj zmiany nadrzędne, w tym nowe produkty, przenoszenie magazynu i ponowne przypisywanie kategorii:

    INSERT INTO products VALUES
  (11, 'Napkin', 'Dining', 'San Francisco'),
  (12, 'Coaster', 'Dining', 'New York');

UPDATE products SET warehouse = 'Los Angeles' WHERE product_id = 1;
UPDATE products SET category = 'Dining' WHERE product_id = 2;

Krok 3. Wykonywanie zapytań dotyczących zestawienia danych zmian

Przed utworzeniem potoku przetwarzania danych warto zapoznać się z nieprzetworzonymi zdarzeniami zmian, aby zrozumieć, co AUTO CDC będzie przetwarzać. Funkcja table_changes() odczytuje dziennik CDF i zwraca każdą przechwyconą operację wraz z kolumnami metadanych:

SELECT
  product_id, product_name, warehouse,
  _change_type, _commit_version
FROM table_changes('products', 1)
ORDER BY _commit_version, product_id;

Na przykład Spoon ma trzy wydarzenia: insert(Seattle), update_preimage(Seattle) i update_postimage(Los Angeles).

Zwróć uwagę, że pojedyncza zmiana logiczna (na przykład przeniesienie łyżki do innego magazynu) powoduje utworzenie wielu zdarzeń: preimage i postimage. W tradycyjnym magazynie napiszesz instrukcję MERGE, aby uzgodnić wszystkie te zdarzenia w tabeli docelowej, zajmować się wstawianiem, aktualizacją i usuwaniem przy użyciu oddzielnej logiki, oraz upewnić się, że zdarzenia są realizowane w odpowiedniej kolejności. Jest to dokładnie złożoność, która AUTO CDC eliminuje w następnym kroku.

Krok 4. Kompilowanie wymiaru typu SCD 2 za pomocą polecenia AUTO CDC

Tabela strumieniowa przetwarza dane przyrostowo. Podczas każdego odświeżania odczytuje tylko nowe wiersze od ostatniego uruchomienia, więc nie musi ponownie przetwarzać pełnego zestawu danych. To sprawia, że dobrze nadaje się do dużych lub często zmieniających się źródeł.

AUTO CDC dodaje przetwarzanie przechwytywania zmian danych w górnej części tabeli przesyłania strumieniowego. Zamiast pisać instrukcję MERGE INTO, która ręcznie obsługuje operacje wstawiania, aktualizacji i usuwania, deklarujesz klucz biznesowy i kolumnę sekwencjonowania, a Azure Databricks stosować poprawną logikę. AUTO CDC automatycznie obsługuje również zdarzenia poza kolejnością, co jest typowym problemem podczas przetwarzania zdarzeń przez MERGE INTO przychodzących z systemów rozproszonych lub obciążeń wsadowych z nakładającymi się znacznikami czasu.

Poniższa instrukcja tworzy tabelę SCD Type 2, która zachowuje pełną historię wersji każdego produktu. Każda wersja otrzymuje __START_AT i __END_AT sygnatury czasowe. Wartość NULL in __END_AT oznacza bieżącą wersję.

CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM (
  SELECT * FROM STREAM products WITH (readChangeFeed = true)
  WHERE _change_type != 'update_preimage'
)
KEYS (product_id)
APPLY AS DELETE WHEN _change_type = 'delete'
SEQUENCE BY _commit_timestamp
COLUMNS * EXCEPT (_change_type, _commit_version, _commit_timestamp)
STORED AS SCD TYPE 2;
  • SCHEDULE REFRESH EVERY 1 DAY: odświeża tabelę zgodnie z harmonogramem dziennym.
  • FLOW AUTO CDC: deklaruje to jako przepływ CDC. Azure Databricks automatycznie stosuje semantyka wstawiania, aktualizowania i usuwania.
  • WHERE _change_type != 'update_preimage': pomija preobrazy. postimage już ma zaktualizowany wiersz.
  • KEYS (product_id): klucz biznesowy. Zdarzenia z tym samym kluczem są scalane z wierszami w wersji.
  • APPLY AS DELETE WHEN _change_type = 'delete': zamyka bieżącą wersję po nadejściu zdarzenia usuwania. Umożliwia to zdefiniowanie warunku identyfikującego zdarzenie usuwania.
  • SEQUENCE BY _commit_timestamp: ustanawia kolejność zdarzeń. Poprawnie obsługuje przyloty poza zamówieniem.
  • STORED AS SCD TYPE 2: zachowuje pełną historię. AUTO CDC obsługuje zarówno typ SCD 1, jak i typ SCD 2.

Zapytaj tabelę wymiarów.

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Łyżka: dwie wersje. Seattle (zamknięte, __END_AT ustawione) i Los Angeles (bieżące, __END_AT = NULL).
  • Fork: dwie wersje. Kategoria Sztućców (zamknięta) i Kategoria Jadalni (bieżąca).
  • Serwetka i podstawka: po jednej wersji (nowo wstawiona, __END_AT = NULL).
  • Wszystkie inne produkty: po jednej wersji każdy (__END_AT = NULL).

Krok 5. Proces usuwania za pośrednictwem potoku

Teraz symuluj dwa wycofane produkty, usuwając je z tabeli źródłowej:

DELETE FROM products WHERE product_id = 9;
DELETE FROM products WHERE product_id = 10;

Te zdarzenia usuwania są rejestrowane w dzienniku CDF, ale tabela strumieniowa jeszcze ich nie zauważyła. Odśwież tabelę przesyłania strumieniowego, aby przetworzyć nowe zdarzenia:

REFRESH STREAMING TABLE products_history;

Wykonaj zapytanie w tabeli wymiarów, aby sprawdzić, czy usunięcia zostały zastosowane.

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;

Bowl i Glass zostały ostatecznie wycofane z __END_AT, oznaczając je jako wycofane. Wszystkie inne bieżące produkty pozostają niezmienione. Tabela przesyłania strumieniowego przetworzyła wyłącznie nowe zdarzenia usuwania, bez ponownego przetwarzania wstawień i aktualizacji z poprzedniego odświeżania.

Krok 6: Zbuduj tabelę faktów z przepływem REPLACE WHERE

Tabela wymiarów w poprzednich krokach używa AUTO CDC, ponieważ śledzi aktualny i historyczny stan każdego produktu. tabele faktów mają różne potrzeby. Tabela faktów rejestruje zdarzenia, takie jak sprzedaż, i zazwyczaj ponownie przetwarza się tylko ostatnie okno, na przykład, aby poprawić spóźnione zapisy. Dla tego wzorca używamy przyrostowego przepływu REPLACE WHERE, który nadpisuje tylko te wiersze odpowiadające predykatowi i pozostawia pozostałe wiersze nietknięte. Pełne porównanie tej konstrukcji z widokami materializowanymi można znaleźć w artykule Kiedy użyć REPLACE WHERE vs. widoki materializowane.

  1. Stwórz tabelę product_sales i załaduj kilka wydarzeń sprzedażowych, niektóre nowsze, inne starsze:

    CREATE OR REPLACE TABLE product_sales (
  sale_id INT,
  product_id INT,
  quantity INT,
  sale_date DATE
);

INSERT INTO product_sales VALUES
  (101, 1, 3, current_date()),
  (102, 2, 1, current_date()),
  (103, 4, 5, date_add(current_date(), -2)),
  (104, 6, 2, date_add(current_date(), -10));

  2. Stwórz tabelę faktów, która oblicza tylko ostatnie 7 dni sprzedaży przy każdym odświeżeniu:

    CREATE OR REFRESH STREAMING TABLE product_sales_recent
SCHEDULE REFRESH EVERY 1 DAY
FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7) BY NAME
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales;
    • FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7): przy każdym odświeżaniu usuwa wiersze w obiekcie docelowym, które odpowiadają predykatowi, ponownie oblicza zapytanie źródłowe dla tego samego zakresu i wstawia wyniki. Wiersze poza 7-dniowym oknem pozostają nietknięte.
    • SCHEDULE REFRESH EVERY 1 DAY: odświeża tabelę według tego samego dziennego harmonogramu co tabela wymiarów.
    • BY NAME: wymagane. Dopasowuje kolumny według nazwy, a nie pozycji.

    Sprzedaż sprzed 10 dni wykracza poza predykat, więc przepływ jej nie ładuje. Utrzymuje się tylko sprzedaż z ostatnich 7 dni.

  3. Wykonaj zapytanie na tabeli faktów:

    SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

    Wynik zawiera trzy sprzedaże z ostatnich 7 dni. Wartości sale_date zależą od dnia, w którym uruchomisz tutorial:

    sale_id product_id ilość data_sprzedaży
    101 1 3 Dziś
    102 2 1 Dziś
    103 4 5 2 dni temu

    Sprzedaż 104 nie jest uwzględniona w wyniku, ponieważ jej data wypada poza 7-dniowym okresem.

Krok 7: Stwórz zagregowany widok materializowany

Teraz, gdy masz tabelę wymiarów, która pozostaje aktualna ze zmianami źródła, możesz dodać warstwę raportowania na górze.

Zmaterializowany widok przechowuje wstępnie obliczone wyniki zapytania jako tabelę fizyczną. W przeciwieństwie do zwykłego widoku, który ponownie wykonuje zapytanie za każdym razem, gdy z niego odczytujesz, zmaterializowany widok utrwala wyniki i tylko ponownie skompiluje wiersze, na które mają wpływ zmiany nadrzędne w każdym odświeżeniu. Dzięki temu jest odpowiedni dla pulpitów nawigacyjnych i raportów, gdzie wydajność zapytań ma znaczenie.

CREATE OR REPLACE MATERIALIZED VIEW products_by_category
SCHEDULE REFRESH EVERY 1 DAY
AS
SELECT
  category,
  COUNT(*) AS active_products
FROM products_history
WHERE __END_AT IS NULL
GROUP BY category;

SCHEDULE REFRESH EVERY 1 DAY oznacza, że ten widok odświeża się zgodnie z harmonogramem dziennym. W połączeniu z tym samym harmonogramem w tabeli strumieniowej masz teraz trzyetapowy potok, w którym zmiany w tabeli źródłowej są przekazywane kaskadowo przez wymiar i docierają do agregatu w każdym cyklu odświeżania. Nie musisz ręcznie odświeżać wymiaru ani agregatu.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Krok 8: Zweryfikuj kaskadę end-to-end

Aby zweryfikować pełną kaskadę pipeline, zmień tabelę źródłową i zarejestruj spóźnioną sprzedaż za wczoraj:

UPDATE products SET warehouse = 'Seattle' WHERE product_id = 3;

INSERT INTO product_sales VALUES (105, 3, 4, date_add(current_date(), -1));

Nóż przenosi się z Denver do Seattle. UPDATE wywołuje pełną kaskadę potoku, pokazując, jak trzy etapy współdziałają:

  1. products Zapisuje zdarzenie zmiany w strumieniu danych zmian.
  2. products_history przetwarza zdarzenie i dodaje nową wersję dla aplikacji Knife.
  3. products_by_category przelicza ponownie tylko dotknięty wiersz Cutlery.

Tabela faktów odświeża się według własnego harmonogramu. Aby uwzględnić teraz opóźnioną sprzedaż zamiast czekać na kolejne zaplanowane odświeżenie, odśwież tabelę faktów ręcznie:

REFRESH STREAMING TABLE product_sales_recent;

Przepływ pracy przelicza ostatnie 7 dni od product_sales, więc sprzedaż 105 pojawia się teraz w product_sales_recent.

Sprawdź:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
WHERE product_id = 3
ORDER BY __START_AT;

SELECT * FROM products_by_category ORDER BY active_products DESC;

SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

Czyszczenie

Aby wyczyścić zasoby utworzone w tym samouczku, użyj następującego kodu SQL:

DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP TABLE IF EXISTS product_sales_recent;
DROP TABLE IF EXISTS product_sales;
DROP TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

Dodatkowe zasoby

  • Ostatnia aktualizacja o