ETL v Databricks SQL

Při práci s velkými objemy dat potřebujete kanál, který může zpracovávat pouze nové a změněné záznamy místo opětovného zpracování celé datové sady. Tomu se říká přírůstkové ETL. V Databricks SQL můžete vytvářet přírůstkové kanály ETL pomocí streamovaných tabulek a materializovaných zobrazení, aniž byste museli psát procedurální kód nebo plánovat ruční aktualizace.

Tento kurz vás provede běžným vzorem: sledování změn produktů v průběhu času. Vytvoříte zdrojovou tabulku, zachytíte události změn, vytvoříte dimenzionální tabulku, která uchovává celou historii každého produktu, udržujete faktovou tabulku, která postupně přepisuje nedávné záznamy, a přidáte agregovanou reportovací vrstvu navrch. Použijete AUTO CDC pro dimenzní tabulku a tok REPLACE WHERE pro tabulku faktů.

Klíčovou funkcí v tomto kurzu je AUTO CDC. V tradičním skladu byste napsali složité MERGE INTO příkazy pro odsouhlasení vložení, aktualizace a odstranění událostí do cílové tabulky. Tento přístup je náchylný k chybám, zejména když události přicházejí mimo pořadí. AUTO CDC se o to stará za vás. Deklarujete obchodní klíč, sloupec sekvencování a to, jestli chcete SCD Type 1 (pouze nejnovější hodnota) nebo SCD Type 2 (úplná historie) a Azure Databricks použije správnou logiku sloučení automaticky. Přehled CDC najdete v tématu Automatizovaná rozhraní API pro CDC: Zjednodušení zachytávání změn údajů pomocí kanálů.

Na konci tohoto kurzu budete mít:

  1. Vytvoření zdrojové tabulky sledující změny pomocí kanálu změn dat.
  2. Prozkoumali nezpracovaná data změn, abyste porozuměli streamu událostí CDC.
  3. Použit AUTO CDC k vytvoření tabulky dimenze typu 2 SCD z těchto událostí.
  4. Zpracování mazacích událostí postupně prostřednictvím pipeline.
  5. Byla vytvořena faktová tabulka, která postupně přepisuje nedávné řádky pomocí flow REPLACE WHERE.
  6. Vytvořili jsme materializované zobrazení, které přírůstkově udržuje agregovanou sestavu.
  7. Nakonfigurováno SCHEDULE REFRESH EVERY 1 DAY tak, aby se změny automaticky šířily prostřednictvím potrubí.

Požadavky

K dokončení tohoto kurzu musíte splnit následující požadavky:

Krok 1: Nastavení katalogu a schématu

Otevřete editor SQL Databricks a nastavte pracovní katalog a schéma. Musíte mít oprávnění k USE katalogu a schématu, které vyberete:

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

Krok 2: Vytvoření zdrojové tabulky a načtení dat

Vytvořte tabulku products s povolenou funkcí Použít informační kanál změn v Azure Databricks (CDF). CDF je funkce Delta Lake, která zaznamenává každou vložení, aktualizaci a odstranění jako dotazovatelný protokol změn. To se podobá datovému proudu CDC z transakčního zdrojového systému, kromě toho, že změny se zaznamenávají přímo v tabulce Delta místo z externího protokolu. CDF zde použijete pro generování událostí změn, které bude využívat následující kanál.

  1. Vytvořte tabulku a načtěte počáteční záznamy:

    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. Simulujte upstreamové změny, včetně nových produktů, přesunu skladu a opětovného přiřazení kategorie:

    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: Dotaz na datový tok změn

Před vytvořením následného kanálu pomáhá podívat se na surové události změn, abyste pochopili, co AUTO CDC bude zpracovávat. Funkce table_changes() čte protokol CDF a vrací všechny zachycené operace spolu se sloupci metadat:

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

Například Spoon má tři události: insert (Seattle), update_preimage (Seattle) a update_postimage (Los Angeles).

Všimněte si, že logická změna (například přesunutí Spoon do jiného skladu) vytváří více událostí: preimage a postimage. V tradičním skladu byste napsali MERGE příkaz pro odsouhlasení všech těchto událostí do cílové tabulky, zpracování vložení, aktualizací a odstranění pomocí samostatné logiky a zajištění, že se události použijí ve správném pořadí. To je přesně složitost, která AUTO CDC eliminuje v dalším kroku.

Krok 4: Sestavení dimenze TYPU 2 SCD pomocí AUTO CDC

Streamovaná tabulka zpracovává data přírůstkově. Při každé aktualizaci čte pouze nové řádky od posledního spuštění, takže nemusí znovu zpracovat úplnou datovou sadu. Díky tomu je vhodná pro velké objemy nebo často se měnící zdroje.

AUTO CDC přidá zpracování zachytávání dat změn nad streamovací tabulku. Místo psaní příkazu MERGE INTO, který ručně zpracovává vkládání, aktualizace a odstranění, deklarujete sloupec obchodního klíče a sekvencování a necháte Azure Databricks použít správnou logiku. AUTO CDC také zpracovává události mimo pořadí automaticky, což je běžný problém při zpracování MERGE INTO událostí přicházejících z distribuovaných systémů nebo dávkových zatížení s překrývajícími se časovými razítky.

Následující příkaz vytvoří tabulku SCD Type 2, která zachovává úplnou historii verzí každého produktu. Každá verze získá __START_AT a __END_AT časové razítko. NULL A __END_AT označuje aktuální verzi.

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: Aktualizuje tabulku podle denního plánu.
  • FLOW AUTO CDC: deklaruje tok CDC. Azure Databricks automaticky uplatňuje sémantiku vložení, aktualizace a odstranění.
  • WHERE _change_type != 'update_preimage': přeskakuje preimage. postimage už má aktualizovaný řádek.
  • KEYS (product_id): obchodní klíč. Události se stejným klíčem jsou sloučeny do verzovaných řádků.
  • APPLY AS DELETE WHEN _change_type = 'delete': Zavře aktuální verzi, když přijde událost odstranění. To vám umožní definovat podmínku, která identifikuje událost odstranění.
  • SEQUENCE BY _commit_timestamp: vytvoří řazení událostí. Zpracovává doručení mimo objednávku správně.
  • STORED AS SCD TYPE 2: uchovává úplnou historii. AUTO CDC podporuje SCD Type 1 i SCD Type 2.

Dotaz na tabulku dimenzí:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Lžička: dvě verze. Seattle (uzavřeno, __END_AT nastavení) a Los Angeles (aktuální, __END_AT = NULL).
  • Fork: dvě verze. Kategorie příborů (uzavřená) a Kategorie Stravování (aktuální).
  • Ubrousek a tácek: jedna verze každého (nově vložená, __END_AT = NULL).
  • Všechny ostatní produkty: každá z nich má jednu verzi (__END_AT = NULL).

Krok 5: Proces odstranění prostřednictvím kanálu

Nyní simulujte dva ukončené produkty jejich odstraněním ze zdrojové tabulky:

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

Tyto události odstranění jsou zaznamenány v protokolu CDF, ale streamovací tabulka je ještě neviděla. Aktualizujte streamovací tabulku, aby zpracovávala nové události:

REFRESH STREAMING TABLE products_history;

Zadejte dotaz na tabulku dimenzí a ověřte, že se použily odstranění:

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

Mísa a sklo jsou nyní označeny jako ukončené pomocí __END_AT sady, což je označuje jako vyřazené. Všechny ostatní aktuální produkty zůstávají beze změny. Streamovaná tabulka zpracovala pouze nové události odstranění bez opětovného zpracování vložení a aktualizací z předchozí aktualizace.

Krok 6: Vytvořte faktovou tabulku s postupem REPLACE WHERE

Tabulka dimenzí v předchozích krocích používá AUTO CDC, protože sleduje aktuální a historický stav každého produktu. Faktové tabulky mají různé požadavky. Faktová tabulka zaznamenává události, jako jsou prodeje, a obvykle znovu zpracováváte pouze nedávné období, například za účelem opravy pozdě doručených záznamů. Pro tento vzor použijte inkrementální tok REPLACE WHERE, který přepíše pouze řádky odpovídající predikátu a všechny ostatní řádky ponechá nedotčené. Pro úplné srovnání této konstrukce s materializovanými pohledy viz Kdy použít REPLACE WHERE vs. materializované pohledy.

  1. Vytvořte tabulku product_sales a načtěte několik událostí prodeje, některé nedávné a některé starší:

    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. Vytvořte faktovou tabulku, která při každém obnovení znovu vypočítá pouze posledních 7 dní prodeje:

    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): při každém obnovení smaže řádky v cíli, které odpovídají predikátu, znovu vypočítá zdrojový dotaz pro stejný rozsah a vloží výsledky. Řádky mimo sedmidenní období zůstávají nedotčené.
    • SCHEDULE REFRESH EVERY 1 DAY: obnovuje tabulku podle stejného denního harmonogramu jako tabulka dimenzí.
    • BY NAME: povinné. Sloupce porovnává podle názvu místo podle pozice.

    Prodej datovaný před 10 dny spadá mimo predikát, takže tok jej nenačte. Udržují se pouze prodeje z posledních 7 dnů.

  3. Dotaz na tabulku faktů:

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

    Výsledek obsahuje tři prodeje za posledních 7 dní. Hodnoty sale_date závisí na dni, kdy tutoriál spustíte:

    sale_id product_id množství datum_prodeje
    101 1 3 Dnes
    102 2 1 Dnes
    103 4 5 před 2 dny

    Prodej 104 není ve výsledku, protože jeho datum je mimo sedmidenní časové okno.

Krok 7: Vytvořte agregovaný materializovaný pohled

Teď, když máte dimenzionální tabulku, která je aktuální se zdrojovými změnami, můžete přidat vrstvu pro tvorbu reportů.

Materializované zobrazení ukládá předem vypočítané výsledky dotazu jako fyzickou tabulku. Na rozdíl od běžného zobrazení, které dotaz znovu spustí při každém čtení z něj, materializované zobrazení zachová výsledky a přepočítá pouze řádky ovlivněné nadřazenými změnami při každé aktualizaci. Díky tomu se dobře hodí pro řídicí panely a sestavy, kde záleží na výkonu dotazů.

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 znamená, že toto zobrazení se aktualizuje podle denního plánu. V kombinaci se stejným plánem ve streamovací tabulce teď máte třífázový kanál, kde dochází ke kaskádovým změnám ve zdrojové tabulce přes dimenzi k agregaci v každém aktualizačním cyklu. Není potřeba ručně obnovovat dimenzi ani agregát.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Krok 8: Ověřte si kompletní kaskádu

Pro ověření celé kaskády pipeline proveďte změnu v zdrojové tabulce a zaznamenejte opožděně dorazivší prodej za včerejšek:

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

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

Nůž se přesune z Denveru do Seattlu. UPDATE spouští celou kaskádu pipeline, která ukazuje, jak tři fáze spolupracují:

  1. products zaznamenává událost změny v kanálu dat změn.
  2. products_history zpracuje událost a přidá novou verzi pro Nůž.
  3. products_by_category přepočítá pouze dotčený řádek Cutlery.

Faktová tabulka se aktualizuje podle vlastního harmonogramu. Chcete-li nyní zpracovat opožděně doručený prodejní záznam místo čekání na další plánované obnovení, ručně obnovte faktovou tabulku:

REFRESH STREAMING TABLE product_sales_recent;

Tok přepočítává posledních 7 dní z product_sales, takže prodej 105 se nyní objevuje v product_sales_recent.

Ověřit:

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;

Vyčištění

K vyčištění prostředků vytvořených v tomto kurzu použijte následující 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;

Dodatečné zdroje

  • Poslední aktualizace v