create_auto_cdc_from_snapshot_flow

Funkce create_auto_cdc_from_snapshot_flow vytvoří tok, který používá kanály Lakeflow ke zpracování zdrojových dat ze snímků databáze funkci záznamu dat (CDC). Podívejte se, jak funguje FUNKCE AUTO CDC FROM SNAPSHOT.

Poznámka:

Tato funkce nahrazuje předchozí funkci apply_changes_from_snapshot(). Obě funkce mají stejný podpis. Databricks doporučuje aktualizovat tak, aby používal nový název.

Důležité

Pro tuto operaci musíte mít cílovou streamovací tabulku. K vytvoření požadované cílové tabulky můžete použít funkci create_streaming_table().

Syntaxe

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Poznámka:

Při AUTO CDC FROM SNAPSHOT zpracování je výchozím chováním vložit nový řádek, pokud v cíli neexistuje odpovídající záznam se stejnými klíči. Pokud existuje odpovídající záznam, aktualizuje se pouze v případě, že se změnily některé hodnoty v řádku. Řádky s klíči, které jsou přítomné v cíli, ale již nejsou ve zdroji, se vymažou.

Další informace o zpracování CDC pomocí snímků najdete v tématu Rozhraní API AUTO CDC: Zjednodušení zachytávání dat změn pomocí datových kanálů. Pro příklady použití této create_auto_cdc_from_snapshot_flow() funkce viz příklady periodického odběru snapshotů, historického odběru snapshotů a SCD typu 1 backfill .

Parametry

Parameter Typ Description
target str Povinné. Název tabulky, která se má aktualizovat. Pomocí funkce create_streaming_table() můžete před spuštěním create_auto_cdc_from_snapshot_flow() funkce vytvořit cílovou tabulku.
source str nebo lambda function Povinné. Název tabulky nebo zobrazení pro pravidelné snímky nebo funkci lambda Pythonu, která vrací datový rámec snímku, který se má zpracovat, a verzi snímku. Viz Implementace argumentusource.
keys list Povinné. Sloupec nebo kombinace sloupců, které jednoznačně identifikují řádek ve zdrojových datech. Slouží k identifikaci událostí CDC, které se vztahují na konkrétní záznamy v cílové tabulce. Můžete zadat jednu z těchto:
  • Seznam řetězců: ["userId", "orderId"]
  • Seznam funkcí Spark SQL col() : [col("userId"), col("orderId"]. Argumenty funkcí col() nemohou obsahovat kvalifikátory. Můžete například použít col(userId), ale nemůžete použít col(source.userId).
stored_as_scd_type str nebo int Určuje, zda se mají ukládat záznamy jako SCD typu 1 nebo SCD typu 2. Nastavte na 1 typ SCD 1 nebo 2 pro SCD typ 2. Výchozí hodnota je SCD typu 1.
track_history_column_list nebo track_history_except_column_list list Podmnožina výstupních sloupců, které se mají sledovat pro historii v cílové tabulce. Slouží track_history_column_list k určení úplného seznamu sloupců, které se mají sledovat. Slouží track_history_except_column_list k určení sloupců, které mají být vyloučeny ze sledování. Můžete deklarovat hodnotu jako seznam řetězců nebo jako funkce Spark SQL col() :
  • track_history_column_list = ["userId", "name", "city"]
  • track_history_column_list = [col("userId"), col("name"), col("city")]
  • track_history_except_column_list = ["operation", "sequenceNum"]
  • track_history_except_column_list = [col("operation"), col("sequenceNum")

Argumenty funkcí col() nemohou obsahovat kvalifikátory. Můžete například použít col(userId), ale nemůžete použít col(source.userId). Výchozí hodnota je zahrnout všechny sloupce v cílové tabulce, pokud funkci nepředáte žádný track_history_column_list argument nebo track_history_except_column_list argument.
once bool Zda spustit snapshot flow jen jednou. Když True, tok je přeskočen během následujících inkrementálních aktualizací po úspěšném commitu. Plné obnovení cíle znovu spustí průběh. Použijte tuto možnost pro jednorázové naložení nebo doplnění snímku. Výchozí hodnota je False.

Notes

Pro SCD Typ 1 cíle v spouštěných pipelinech může jeden create_auto_cdc_from_snapshot_flow() tok sdílet cíl s jedním nebo více AUTO CDC toky definovanými v Python nebo SQL. Platí následující požadavky:

  • Dej každému AUTO CDC toku jedinečné jméno.
  • Používejte stejný počet klíčů ve stejném pořadí pro všechny toky. Názvy klíčů typu snapshot-flow se porovnávají podle pádového necitlivosti s AUTO CDC názvy klíčů. Více AUTO CDC toků musí používat stejná jména klíčů a pouzdro.
  • Použijte přesně stejný datový typ pro verzi snapshotu a sloupec sekvenace každého AUTO CDC toku.
  • Neurčujte očekávání během procesu AUTO CDC FROM SNAPSHOT .
  • Nepoužívejte IGNORE NULL UPDATES. V Python nenastavujte ignore_null_updates, ignore_null_updates_column_list, ani ignore_null_updates_except_column_list na create_auto_cdc_flow() flow.
  • Pokud přidáte AUTO CDC tok do existujícího AUTO CDC FROM SNAPSHOT cíle, cíl nesmí obsahovat uživatelské sloupce, jejichž názvy jsou v rozporu s AUTO CDC rezervovanými systémovými sloupci.
  • Tento vzor nepoužívejte pro SCD typu 2 nebo bitemporální cíle.

Například viz Přidat zpětný záplň do tabulky typu 1 AUTO CDC SCD.

Implementace argumentu source

Funkce create_auto_cdc_from_snapshot_flow() obsahuje source argument. Pro zpracování historických snímků se očekává, že argumentem je funkce lambda Pythonu, která vrací dvě hodnoty source funkce: datový rámec Pythonu obsahující data snímku, create_auto_cdc_from_snapshot_flow() která se mají zpracovat, a verzi snímku.

První volání funkce musí vrátit snapshot DataFrame a verzi. Pokud první volání vrátí None, aktualizace pipeline selže. Po zpracování alespoň jednoho snímku se return None označí, že nejsou k dispozici žádné další snímky.

Následuje podpis funkce lambda:

lambda Any => Optional[(DataFrame, Any)]
  • Argumentem funkce lambda je nejnovější zpracovaná verze snímku.
  • Návratová hodnota funkce lambda je None nebo řazená kolekce členů dvou hodnot: První hodnota řazené kolekce členů je datový rámec obsahující snímek, který se má zpracovat. Druhá hodnota n-tice je verze snímku, která udává logické pořadí snímku.

Příklad, který implementuje a volá funkci lambda:

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Modul runtime kanálů Lakeflow provádí následující kroky pokaždé, když se aktivuje kanál obsahující create_auto_cdc_from_snapshot_flow() funkci:

  1. next_snapshot_and_version Spustí funkci, která načte další datový rámec snímku a odpovídající verzi snímku.
  2. Pokud první volání nevrátí žádný DataFrame, aktualizace selže. Pokud pozdější volání nevrátí žádný DataFrame, běh končí a aktualizace pipeline je označena jako dokončená.
  3. Rozpozná změny v novém snímku a postupně je použije na cílovou tabulku.
  4. Vrací se ke kroku č. 1, aby načetl další snímek a jeho verzi.
  • Poslední aktualizace v