Replikace externí tabulky RDBMS pomocí AUTO CDC

Tabulku z externího systému pro správu relačních databází (RDBMS) můžete do Azure Databricks replikovat pomocí rozhraní API AUTO CDC v pipelines. Naučíte se:

  • Běžné vzory pro nastavení zdrojů
  • Jak provést jednorázovou úplnou kopii existujících dat pomocí once toku.
  • Jak průběžně zpracovávat nové změny pomocí change toku.

Tento model je ideální pro vytváření pomalu se měnících tabulek dimenzí (SCD) nebo udržování cílové tabulky synchronizované s externím systémem záznamů.

Než začnete

V této příručce se předpokládá, že máte přístup k následujícím datovým sadám ze zdroje:

  • Úplný snímek zdrojové tabulky v cloudovém úložišti Tato datová sada se používá pro počáteční načtení.
  • Kanál průběžných změn naplněný do stejného umístění cloudového úložiště (například pomocí Debezium, Kafka nebo CDC založeného na protokolech). Tento informační kanál je vstupem pro probíhající AUTO CDC proces.

Nastavení zdrojových zobrazení

Nejprve definujte dvě zdrojová zobrazení pro naplnění cílové tabulky rdbms_orders z cesty cloudového úložiště orders_snapshot_path. Obě jsou vytvořena jako streamovací pohledy nad nezpracovanými daty v cloudovém úložišti. Použití zobrazení poskytuje vyšší efektivitu, protože data nemusí být před použitím v AUTO CDC procesu zapsána.

  • První zobrazení zdroje je úplný snímek (full_orders_snapshot)
  • Druhým je průběžný kanál změn (rdbms_orders_change_feed).

Příklady v této příručce používají cloudové úložiště jako zdroj, ale můžete použít libovolný zdroj podporovaný tabulkami streamování.

full_orders_snapshot()

Tento krok vytvoří kanál se zobrazením, které načte počáteční úplný snímek dat objednávek.

Python

Následující příklad Pythonu:

  • Použití spark.readStream s automatickým zavaděčem (format("cloudFiles"))
  • Čte soubory JSON z adresáře definovaného pomocí orders_snapshot_path
  • Nastaví includeExistingFiles na true tak, aby se zajistilo zpracování historických dat, která již v cestě existují.
  • Nastaví inferColumnTypes na true pro automatické odvození schématu.
  • Vrátí všechny sloupce s .select("\*")
@dp.view()
def full_orders_snapshot():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.includeExistingFiles", "true")
        .option("cloudFiles.inferColumnTypes", "true")
        .load(orders_snapshot_path)
        .select("*")
    )

SQL

Následující příklad SQL předá možnosti jako mapu řetězcových párů klíč-hodnota. orders_snapshot_path by měla být k dispozici jako proměnná SQL (například definovaná pomocí parametrů kanálu nebo ručně interpolovaná).

CREATE OR REFRESH VIEW full_orders_snapshot
AS SELECT *
FROM STREAM read_files("${orders_snapshot_path}", "json", map(
  "cloudFiles.includeExistingFiles", "true",
  "cloudFiles.inferColumnTypes", "true"
));

rdbms_orders_change_feed()

Tento krok vytvoří druhé zobrazení, které čte data přírůstkové změny (například z protokolů CDC nebo tabulek změn). Čte z orders_cdc_path a předpokládá, že soubory JSON ve stylu CDC se do této cesty pravidelně vkládají.

Python

@dp.view()
def rdbms_orders_change_feed():
return (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.option("cloudFiles.includeExistingFiles", "true")
.option("cloudFiles.inferColumnTypes", "true")
.load(orders_cdc_path)

SQL

V následujícím příkladu SQL je ${orders_cdc_path} proměnná a dá se interpolovat nastavením hodnoty v nastaveních kanálu nebo explicitním nastavením proměnné v kódu.

CREATE OR REFRESH VIEW rdbms_orders_change_feed
AS SELECT *
FROM STREAM read_files("${orders_cdc_path}", "json", map(
"cloudFiles.includeExistingFiles", "true",
"cloudFiles.inferColumnTypes", "true"
));

Počáteční hydrace (jednorázový tok)

Teď, když jsou zdroje nastavené, AUTO CDC logika sloučí oba zdroje do cílové tabulky streamování. Nejprve pomocí jednorázového AUTO CDC toku ONCE=TRUE zkopírujte celý obsah tabulky RDBMS do streamované tabulky. Tím se cílová tabulka připraví s historickými daty, aniž by byla v budoucích aktualizacích opětovně použita.

Python

from pyspark import pipelines as dp

# Step 1: Create the target streaming table

dp.create_streaming_table("rdbms_orders")

# Step 2: Once Flow — Load initial snapshot of full RDBMS table

dp.create_auto_cdc_flow(
  flow_name = "initial_load_orders",
  once = True,  # one-time load
  target = "rdbms_orders",
  source = "full_orders_snapshot",  # e.g., ingested from JDBC into bronze
  keys = ["order_id"],
  sequence_by = "timestamp",
  stored_as_scd_type = "1"
)

SQL


-- Step 1: Create the target streaming table
CREATE OR REFRESH STREAMING TABLE rdbms_orders;

-- Step 2: Once Flow for initial snapshot
CREATE FLOW rdbms_orders_hydrate
AS AUTO CDC ONCE INTO rdbms_orders
FROM stream(full_orders_snapshot)
KEYS (order_id)
SEQUENCE BY timestamp
STORED AS SCD TYPE 1;

Tok once se spustí pouze jednou. Nové soubory, které se přidají do full_orders_snapshot po vytvoření kanálu, se ignorují.

Důležité

Provedení úplné aktualizace rdbms_orders streamované tabulky opětovně spustí once tok. Pokud dojde k odebrání počátečních dat snímku v cloudovém úložišti, dojde ke ztrátě dat.

Kanál průběžných změn (informační tok změn)

Po počátečním načtení snímku použijte jiný AUTO CDC tok k průběžnému zpracování změn z kanálu CDC systému pro správu RDBMS. Díky tomu bude rdbms_orders vaše tabulka aktuální díky vkládáním, aktualizacím a odstraňováním.

Python

from pyspark import pipelines as dp

# Step 3: Change Flow — Ingest ongoing CDC stream from source system

dp.create_auto_cdc_flow(
flow_name = "orders_incremental_cdc",
target = "rdbms_orders",
source = "rdbms_orders_change_feed", # e.g., ingested from Kafka or Debezium
keys = ["order_id"],
sequence_by = "timestamp",
stored_as_scd_type = "1"
)

SQL

-- Step 3: Continuous CDC ingestion
CREATE FLOW rdbms_orders_continuous
AS AUTO CDC INTO rdbms_orders
FROM stream(rdbms_orders_change_feed)
KEYS (order_id)
SEQUENCE BY timestamp
STORED AS SCD TYPE 1;

Úvahy

Idempotence doplňování Tok once se znovu spustí, pouze když se cílová tabulka plně aktualizuje.
Více toků Pomocí několika toků změn můžete sloučit opravy, pozdě přicházející data nebo alternativní informační kanály, ale všechny musí sdílet stejné schéma a klíče.
Úplná aktualizace Úplná aktualizace rdbms_orders streamované tabulky opět spustí once proces. To může vést ke ztrátě dat, pokud počáteční umístění cloudového úložiště vyřízlo počáteční data snímku.
Pořadí provádění toku Pořadí provádění toku není důležité. Konečný výsledek je stejný.

Dodatečné zdroje

  • Last updated on