ETL az Databricks SQL-ben

Nagy mennyiségű adat kezelésekor olyan folyamatra van szükség, amely a teljes adathalmaz újrafeldolgozása helyett csak az új és módosított rekordokat tudja feldolgozni. Ezt növekményes ETL-nek nevezzük. A Databricks SQL-ben növekményes ETL-folyamatokat hozhat létre streamelő táblák és materializált nézetek használatával, eljárási kód írása vagy manuális frissítések ütemezése nélkül.

Ez az oktatóanyag végigvezeti egy gyakori mintán: nyomon követheti a termék változásait az idő függvényében. Létrehozhat egy forrástáblát, rögzítheti a változási eseményeket, létrehozhat egy dimenziótáblát, amely megőrzi az egyes termékek teljes előzményeit, és felül egy összesített jelentési réteget vehet fel.

Ebben az oktatóanyagban a legfontosabb funkció a AUTO CDC. Egy hagyományos raktárban összetett MERGE INTO utasításokat kell írnia, hogy összehangolja a beszúrási, frissítési és törlési eseményeket egy céltáblában. Ez a megközelítés hibára hajlamos, különösen akkor, ha az események nem sorrendben érkeznek. AUTO CDC ezt kezeli neked. Deklarálja az üzleti kulcsot, a szekvenálási oszlopot, valamint azt, hogy az SCD 1. típusú (csak a legújabb érték) vagy az SCD 2. típusú (teljes előzmény) típust szeretné-e használni, és Azure Databricks automatikusan alkalmazza a helyes egyesítési logikát. A CDC áttekintéséhez tekintse meg az AUTO CDC API-kat: Egyszerűsítse az adatváltozás rögzítését a folyamatokkal.

Az oktatóanyag végére a következőkre lesz szüksége:

1. Létrehozott egy forrástáblát, amely nyomon követi az adatcsatorna módosításával kapcsolatos változásokat.
2. Megvizsgálta a nyers módosítási adatokat a CDC eseménystreamjének megértéséhez.
3. SCD AUTO CDC 2. típusú dimenziótáblát épített ezekből az eseményekből.
4. Törlési események fokozatos feldolgozása a folyamaton keresztül.
5. Olyan materializált nézetet hozott létre, amely növekményesen tartja karban az összesítő jelentést.
6. Konfigurálva lett SCHEDULE REFRESH EVERY 1 DAY, hogy a változtatások automatikusan átmenjenek a csővezetéken.

Követelmények

Az oktatóanyag elvégzéséhez meg kell felelnie a következő követelményeknek:

• Egy Azure Databricks munkaterület, amelyen engedélyezve van a Unity Catalog.
• SQL-raktár (kiszolgáló nélküli vagy pro).
• Rendelkezik engedéllyel számítási erőforrás létrehozására vagy egy számítási erőforráshoz való hozzáférésre.
• A kiszolgáló nélküli számítás engedélyezve van a fiókjához. Tekintse meg a korlátozott regionális rendelkezésre állású funkciókat.

1. lépés: A katalógus és a séma beállítása

Nyissa meg a Databricks SQL-szerkesztőt , és állítsa be a munkakatalógust és a sémát. Engedéllyel kell rendelkeznie a kiválasztott katalógushoz USE és sémához:

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

2. lépés: Forrástábla létrehozása és adatok betöltése

Hozzon létre egy olyan products táblát, amelyen engedélyezve van a Változásadatok hírcsatornájának használata az Azure Databricksben (CDF). A CDF egy Delta Lake-funkció, amely minden beszúrást, frissítést és törlést lekérdezhető változásnaplóként rögzít. Ez hasonló a tranzakciós forrásrendszerből származó CDC-adatfolyamhoz, kivéve, ha a módosítások közvetlenül a Delta táblában vannak rögzítve, nem pedig egy külső naplóból. Itt a CDF használatával hozhatja létre az alsóbb rétegbeli folyamat által használt változási eseményeket.

1. Hozza létre a táblát, és töltse be a kezdeti rekordokat:

   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. A felsőbb rétegbeli változások szimulálása, beleértve az új termékeket, a raktár áthelyezését és a kategória-hozzárendelést:

   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;
   

3. lépés: A változásadatcsatorna lekérdezése

A lefolyásirányú csővezeték építése előtt segít áttekinteni a nyers változási eseményeket, hogy megértsük, mit fog feldolgozni a AUTO CDC. A table_changes() függvény beolvassa a CDF-naplót, és visszaadja az összes rögzített műveletet a metaadatoszlopokkal együtt:

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

A Kanálnak például három eseménye van: egy insert (Seattle), egy update_preimage (Seattle) és egy update_postimage (Los Angeles).

Figyelje meg, hogy egyetlen logikai változás (például a Kanál áthelyezése egy másik raktárba) több eseményt eredményez: egy előképet és egy utóképet. Egy hagyományos raktárban egy utasítást MERGE kell írnia, amely az összes eseményt egy céltáblába alakítja, külön logikával kezeli a beszúrásokat, frissítéseket és törléseket, és gondoskodik arról, hogy az események a megfelelő sorrendben legyenek alkalmazva. Ez pontosan az összetettség, amely AUTO CDC kiküszöböli a következő lépésben.

4. lépés: SCD 2-es típusú dimenzió létrehozása a következővel: AUTO CDC

A streamelő tábla növekményesen dolgozza fel az adatokat. Minden frissítéskor csak az új sorokat olvassa be az utolsó futtatás óta, így nem kell újra feldolgoznia a teljes adatkészletet. Így kiválóan alkalmas nagy mennyiségű vagy gyakran változó forrásokhoz.

AUTO CDC A módosítási adatrögzítési feldolgozást egy streamelési tábla tetején adja hozzá. A beszúrásokat, frissítéseket és törléseket manuálisan kezelő MERGE INTO utasítás írása helyett deklarálhatja az üzleti kulcsot és a szekvenálási oszlopot, és Azure Databricks alkalmazhatja a helyes logikát. AUTO CDC automatikusan kezeli a sorrenden kívüli eseményeket is, ami gyakori probléma, amikor elosztott rendszerekből vagy átfedésben lévő időbélyegekkel rendelkező kötegelt terhelésekből érkező eseményeket kezelünk MERGE INTO segítségével.

Az alábbi utasítás létrehoz egy 2. típusú SCD-táblát, amely megőrzi az egyes termékek teljes verzióelőzményeit. Minden verzió kap __START_AT és __END_AT időbélyegeket. Az aktuális verziót a NULL__END_AT jelöli.

CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM STREAM products WITH (readChangeFeed = true)
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: napi ütemezés szerint frissíti a táblát.
• FLOW AUTO CDC: ezt CDC-folyamatként deklarálja. Azure Databricks automatikusan alkalmazza a szemantikák beszúrását, frissítését és törlését.
• KEYS (product_id): az üzleti kulcs. Az azonos kulccsal rendelkező események verziószámozott sorokba vannak egyesítve.
• APPLY AS DELETE WHEN _change_type = 'delete': a törlési esemény érkezésekor bezárja az aktuális verziót. Ez lehetővé teszi a törlési eseményt azonosító feltétel meghatározását.
• SEQUENCE BY _commit_timestamp: eseményrendezést hoz létre. Megfelelően kezeli a rendelésen kívüli érkezéseket.
• STORED AS SCD TYPE 2: megtartja a teljes előzményt. AUTO CDC Támogatja az SCD 1. és 2. típusú SCD-t is.

A dimenziótábla lekérdezése:

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

• Kanál: két verzió. Seattle (zárt, __END_AT set) és Los Angeles (jelenlegi, __END_AT = NULL).
• Elágazás: két verzió. Evőeszközök kategóriája (zárt) és Étkezési kategória (aktuális).
• Szalvéta és poháralátét: egy-egy verzió (újonnan beszúrt, __END_AT = NULL).
• Minden más termék: mindegyik egy-egy verzió (__END_AT = NULL).

5. lépés: A törlések feldolgozása a csatornán keresztül

Most szimuláljon két megszűnt terméket a forrástáblából való törléssel:

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

Ezek a törlési események a CDF-naplóban vannak rögzítve, de a streamelési tábla még nem látta őket. Frissítse a streamelési táblát az új események feldolgozásához:

REFRESH STREAMING TABLE products_history;

A dimenziótábla lekérdezése a törlések alkalmazásának ellenőrzéséhez:

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

A tálat és az üveget most kivonjuk a forgalomból a __END_AT készlet részeként, és megszüntetettként jelöljük őket. Az összes többi jelenlegi termék változatlan marad. A streamelő tábla csak az új törlési eseményeket dolgozza fel az előző frissítés beszúrásainak és frissítéseinek újrafeldolgozása nélkül.

6. lépés: Összesített materializált nézet létrehozása

Most, hogy rendelkezik egy olyan dimenziótáblával, amely naprakész marad a forrásváltozásokkal, felvehet egy jelentési réteget a tetejére.

A materializált nézet fizikai táblaként tárolja az előre kiszámított lekérdezési eredményeket. A normál nézettől eltérően, amely minden olvasáskor újra végrehajtja a lekérdezést, a materializált nézet megőrzi az eredményeket, és csak az egyes frissítések felsőbb rétegbeli módosításai által érintett sorokat újrafordítja. Ez jól használható olyan irányítópultokhoz és jelentésekhez, ahol a lekérdezési teljesítmény számít.

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 azt jelenti, hogy ez a nézet napi ütemezés szerint frissül. Az adatfolyam táblának ugyanaz az ütemezése áll rendelkezésre, így most egy háromfázisú adatfolyam feldolgozási lánccal rendelkezik, amelyben a forrástábla módosításai kaszkádként továbbítódnak a dimenziókon keresztül az összesített adatokba minden frissítési ciklus során. Nincs manuálisan futtatható frissítés.

SELECT * FROM products_by_category ORDER BY active_products DESC;

7. lépés: A végpontok közötti kaszkád ellenőrzése

A teljes pipelénekaszkád ellenőrzéséhez módosítsa a forrástáblát:

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

A kés Denverből Seattle-be költözik. Ez az egyetlen DML-módosítás aktiválja a teljes folyamat kaszkádolását, bemutatva a három fázis együttes működését:

1. products cdf-en keresztül rögzíti a változási eseményt.
2. products_history feldolgozza az eseményt, és hozzáad egy új verziót a Késhez.
3. products_by_category csak az érintett Cutlery sort újraszámítja.

Ellenőrizze:

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;

Takarítás

Az oktatóanyag által létrehozott erőforrások törléséhez használja a következő SQL-t:

DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP STREAMING TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

További erőforrások

• adatfolyam táblák
• Materializált nézetek
• CREATE STREAMING TABLE ... FLOW AUTO CDC
• Az AUTO CDC API-k: A változási adatok rögzítésének egyszerűsítése csatornákkal
• Spark deklaratív adatfolyamok