Die AUTO CDC-APIs: Vereinfachen der Änderungsdatenerfassung mit Pipelines

Lakeflow-Pipelines vereinfachen die Änderungsdatenerfassung (Change Data Capture, CDC) mit den AUTO CDC und AUTO CDC FROM SNAPSHOT APIs. Diese APIs automatisieren die Komplexität der Berechnung von sich langsam ändernden Dimensionen (Slowly Changing Dimensions, SCD) vom Typ 1 und Typ 2 aus einem CDC-Feed oder Datenbankmomentaufnahmen. Die AUTO CDC API unterstützt auch die bitemporale Nachverfolgung (Beta), bei der Änderungen über zwei Zeitdimensionen hinweg erfasst werden. Weitere Informationen zu SCD Typ 1 und Typ 2 finden Sie unter Change Data Capture und Snapshots. Weitere Informationen zur bitemporalen Nachverfolgung finden Sie unter Bitemporal AUTO CDC.

Hinweis

Die AUTO CDC APIs ersetzen die APPLY CHANGES APIs und weisen dieselbe Syntax auf. Die APPLY CHANGES APIs sind weiterhin verfügbar, aber Databricks empfiehlt die Verwendung der AUTO CDC APIs stattdessen.

Die verwendete API hängt von der Quelle Ihrer Änderungsdaten ab:

  • AUTO CDC: Verwenden Sie diese Option, wenn für die Quelldatenbank ein CDC-Feed aktiviert ist. AUTO CDC verarbeitet Änderungen aus einem Änderungsdatenfeed (CDF). Sie wird sowohl in den SQL- als auch in den Python-Schnittstellen der Pipeline unterstützt.
  • AUTO CDC FROM SNAPSHOT: Verwenden Sie diese Option, wenn CDC nicht in der Quelldatenbank aktiviert ist und nur Momentaufnahmen verfügbar sind. Diese API vergleicht Momentaufnahmen, um Änderungen zu bestimmen und diese dann zu verarbeiten. Sie wird sowohl in den SQL- als auch in den Python-Schnittstellen der Pipeline unterstützt.

Für SCD Typ-1-Ziele kann man auch einen AUTO CDC FROM SNAPSHOT Flow mit einem oder mehreren AUTO CDC Flows kombinieren. Verwenden Sie dieses Muster, wenn eine Quelle periodische autoritative Snapshots und einen CDC-Feed mit geringerer Latenz zwischen den Snapshots bereitstellt.

Beide APIs unterstützen das Aktualisieren von Tabellen mit SCD-Typ 1 und Typ 2:

  • Verwenden Sie SCD Type 1, um Datensätze direkt zu aktualisieren. Der Verlauf wird für aktualisierte Datensätze nicht beibehalten.
  • Verwenden Sie SCD Type 2, um einen Verlauf von Datensätzen beizubehalten, entweder für alle Aktualisierungen oder für Aktualisierungen an einer bestimmten Gruppe von Spalten.

Nur für AUTO CDC können Sie auch bitemporalen Speicher verwenden, der den SCD-Typ-2-Verlauf erweitert, um Änderungen über zwei Zeitdimensionen hinweg nachzuverfolgen: Geschäftszeit und Systemzeit. Bitemporal befindet sich in der Beta-Phase. Siehe Bitemporal AUTO CDC.

AUTO CDC unterstützt auch partielle Aktualisierungen, bei denen ein Änderungsdatensatz nur eine Teilmenge von Spalten aktualisiert. Siehe "Teilaktualisierungen anwenden".

Die AUTO CDC APIs werden von Apache Spark Declarative Pipelines nicht unterstützt.

Informationen zur Syntax und weitere Referenzen finden Sie unter AUTO CDC INTO (Pipelines), create_auto_cdc_flow und create_auto_cdc_from_snapshot_flow.

Hinweis

Nutzen Sie diese APIs, um Tabellen in Ihren Pipelines basierend auf Änderungen der Quelldaten zu aktualisieren. Informationen zum Aufzeichnen und Abfragen von Änderungsinformationen auf Zeilenebene für Delta-Tabellen finden Sie unter Verwenden des Änderungsdatenfeeds für Azure Databricks.

Anforderungen

Um die CDC-APIs nutzen zu können, muss Ihre Pipeline so konfiguriert sein, dass sie serverlose Lakeflow-Pipelines oder die Lakeflow-Pipelines der Pro- oder Advanced-Versionen verwendet.

Funktionsweise von AUTO CDC

Um die CDC-Verarbeitung mit AUTO CDCdurchzuführen, erstellen Sie eine Streamingtabelle, und verwenden Sie dann die AUTO CDC ... INTO Anweisung in SQL oder die create_auto_cdc_flow() Funktion in Python, um die Quelle, Schlüssel und Sequenzierung für den Änderungsfeed anzugeben. Eine Erläuterung der Funktionsweise von Sequenzierungs- und SCD-Logik finden Sie unter Ändern der Datenerfassung und Momentaufnahmen. Weitere Informationen finden Sie in den Auto CDC-Beispielen.

Verwenden Sie für die anfängliche Hydratisierung aus einer Quelle mit einem Änderungsfeed AUTO CDC mit einem once-Flow, und setzen Sie dann die Verarbeitung des Änderungsfeeds fort. Siehe Replizieren einer externen RDBMS-Tabelle mit AUTO CDC.

Ausführliche Informationen zur Syntax finden Sie unter AUTO CDC INTO (Pipelines) oder create_auto_cdc_flow.

Funktionsweise von AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT ermittelt Änderungen in Quelldaten durch den Vergleich von Momentaufnahmen in der richtigen Reihenfolge. Sie können Momentaufnahmen direkt aus einer Delta-Tabelle, aus Cloudspeicherdateien oder über JDBC lesen. Sie wird sowohl in den SQL- als auch in den Python-Schnittstellen der Pipeline unterstützt.

Um die CDC-Verarbeitung mit AUTO CDC FROM SNAPSHOTdurchzuführen, erstellen Sie eine Streaming-Tabelle und definieren Sie dann den Fluss:

  • In SQL verwenden Sie die AUTO CDC ... FROM SNAPSHOT Form der CREATE FLOW Anweisung oder der eingebetteten FLOW AUTO CDC Klausel von CREATE STREAMING TABLE. Die Quelle ist eine erforderliche FROM SNAPSHOT (snapshot_query) Klausel und eine optionale WITH VERSION (version_query) Klausel, die die nächste Snapshot-Version auswählt, die verarbeitet wird. Siehe CREATE FLOW (pipelines).
  • In Python verwenden Sie die create_auto_cdc_from_snapshot_flow() Funktion, um den Snapshot, die Schlüssel und andere Argumente anzugeben. Siehe create_auto_cdc_from_snapshot_flow.

Ausführliche Informationen zu den beiden Ingestionsmustern und deren Verwendung finden Sie unter Snapshot-Verarbeitungsmuster. Sehen Sie sich die AUTO CDC FROM SNAPSHOT-Beispiele an.

AUTO CDC FROM SNAPSHOT und AUTO CDC kombinieren

Die Snapshot-Version und die CDC-Sequenzspalte bilden eine Ordnungsdomäne. Ein CDC-Ereignis, das neuer als das neueste Snapshot ist, hat Vorrang gegenüber dem Snapshot. Ein neueres Snapshot hat Vorrang gegenüber älteren CDC-Ereignissen und behandelt Schlüssel, die im Snapshot fehlen, als gelöscht in der Snapshot-Version. Neuere CDC-Ereignisse können diese Schlüssel erhalten oder wiederherstellen.

Ein einheitliches Ziel muss folgende Anforderungen erfüllen:

  • Das Ziel verwendet SCD Type 1.
  • Das Ziel hat genau einen AUTO CDC FROM SNAPSHOT Flow und einen oder mehrere AUTO CDC Flows.
  • Jeder AUTO CDC Fluss hat einen einzigartigen Namen.
  • Alle Flows verwenden die gleiche Anzahl von Schlüsseln in derselben Reihenfolge. Snapshot-Flow-Schlüsselnamen werden groß- und kleinschreibungsunsensitiv mit AUTO CDC Schlüsselnamen verglichen. Mehrere Flows AUTO CDC müssen identische Schlüsselnamen und Groß- und Kleinschreibung verwenden.
  • Die Snapshot-Version und jede CDC-Sequenzspalte haben exakt denselben Datentyp.
  • Der Flow AUTO CDC FROM SNAPSHOT definiert keine Erwartungen.
  • Die Flows AUTO CDC verwenden weder IGNORE NULL UPDATES noch dessen Spaltenlistenvarianten.
  • Die Pipeline verwendet den Trigger-Modus.
  • Wenn Sie einen AUTO CDC FROM SNAPSHOT Fluss zu einem bestehenden AUTO CDC Ziel hinzufügen, darf das Ziel keine Benutzerspalten enthalten, deren Namen mit reservierten AUTO CDC Systemspalten kollidieren.

Sowohl AUTO CDC FROM SNAPSHOT- als auch AUTO CDC-Flows können die Python- oder SQL-Pipeline-Schnittstelle verwenden. Du kannst SQL- und Python-Flows im selben Ziel mischen.

Verwenden Sie once=True in Python oder ONCE in SQL im AUTO CDC FROM SNAPSHOT Flow, um einen Snapshot einmalig nachzufüllen, während die AUTO CDC Flows neue Ereignisse bei späteren Updates verarbeiten. Eine vollständige Aktualisierung des Ziels führt den einmaligen Snapshot-Fluss erneut aus. Ein vollständiges Beispiel finden Sie unter Hinzufügen einer Auffüllung zu einer AUTO CDC SCD Type 1-Tabelle.

Dieses Muster unterstützt keine SCD Typ 2 oder bitemporale Ziele.

Verwenden mehrerer Spalten für die Sequenzierung

Um nach mehreren Spalten zu sequenzieren (z. B. ein Zeitstempel und eine ID zum Unterbrechen von Bindungen), verwenden Sie eine STRUCT , um sie zu kombinieren. Die API sortiert zuerst nach dem ersten Feld, und falls es einen Gleichstand gibt, nach dem zweiten Feld usw.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

sequence_by = struct("timestamp_col", "id_col")

AUTO CDC-Beispiele

Die folgenden Beispiele veranschaulichen die SCD-Typ 1- und Typ 2-Verarbeitung mithilfe einer Änderungsdatenfeedquelle. Die Beispieldaten erstellen neue Benutzerdatensätze, löschen einen Benutzerdatensatz und aktualisieren Benutzerdatensätze. Im SCD-Typ-1-Beispiel treffen die letzten UPDATE-Vorgänge verspätet ein und werden aus der Zieltabelle gelöscht, was die Behandlung von Ereignissen außerhalb der Reihenfolge veranschaulicht.

Im Folgenden sind die Eingabedatensätze aufgeführt, die in diesen Beispielen verwendet werden. Diese Daten werden durch Ausführen der Abfrage im Abschnitt "Beispieldaten erstellen " erstellt.

userId name city operation Sequenznummer
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lilie Cancun INSERT 2
123 NULL NULL Löschen 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Wenn Sie die Kommentierung der letzten Zeile in der Beispieldatengenerierungsabfrage aufheben, wird der folgende Datensatz eingefügt, der angibt, dass die Tabelle geleert werden soll (löschen Sie die Tabelle bei sequenceNum=3)

userId name city operation Sequenznummer
NULL NULL NULL TRUNCATE 3

Hinweis

Alle folgenden Beispiele enthalten Optionen zum Angeben von Beiden DELETE und TRUNCATE Vorgängen, aber jede ist optional.

Erstellen von Beispieldaten

Führen Sie die folgenden Anweisungen aus, um ein Beispiel-Dataset zu erstellen. Dieser Code soll nicht als Teil einer Pipelinedefinition ausgeführt werden. Führen Sie dieses aus dem Explorationsordner Ihrer Pipeline aus, anstatt aus dem Transformationsordner.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

SCD-Typ 1-Updates verarbeiten

SCD Type 1 behält nur die neueste Version jedes Datensatzes bei. Im folgenden Beispiel wird aus dem oben erstellten Änderungsdatenfeed gelesen und Änderungen auf ein Streamingtabellenziel angewendet. Du brauchst eine Pipeline, um diesen Code auszuführen. Siehe Entwickeln und Debuggen von ETL-Pipelines mit dem Lakeflow Pipelines-Editor.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Nach dem Ausführen des SCD-Typs 1-Beispiels enthält die Zieltabelle die folgenden Datensätze:

userId name city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lilie Cancun

Benutzer 123 (Isabel) wurde gelöscht und nicht angezeigt. Benutzer 125 (Mercedes) zeigt nur die neueste Stadt (Guadalajara) an, da SCD Typ 1 vorherige Werte überschreibt. Die frühere UPDATE bei sequenceNum=5 wurde entfernt, da ein späteres Update bei sequenceNum=6 ankam.

Nach dem Ausführen des Beispiels mit dem auskommentierten TRUNCATE-Datensatz wird die Tabelle bei sequenceNum=3 gelöscht. Dies bedeutet, dass Datensätze 124 und 126 nicht in der Tabelle enthalten sind und die endgültige Zieltabelle nur den folgenden Datensatz enthält:

userId name city
125 Mercedes Guadalajara

Verarbeitung von SCD-Typ-2-Updates

SCD Typ 2 behält einen vollständigen Verlauf der Änderungen bei, indem neue Zeilen für jede Version eines Datensatzes erstellt werden, wobei die __START_AT- und __END_AT-Spalten angeben, wann jede Version aktiv war.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Nach dem Ausführen des SCD-Typs 2-Beispiels enthält die Zieltabelle die folgenden Datensätze:

userId name city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 NULL
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 NULL
126 Lilie Cancun 2 NULL

Die Tabelle speichert den kompletten Verlauf. Der Benutzer 123 hat zwei Versionen (beendet bei Sequenz 6, wenn er gelöscht wurde). Benutzer 125 verfügt über drei Versionen mit Stadtänderungen. Datensätze mit __END_AT = null sind derzeit aktiv.

Nachverfolgen einer Spaltenteilmenge mit SCD Typ 2

Standardmäßig erstellt SCD Type 2 immer dann eine neue Version, wenn sich ein Spaltenwert ändert. Sie können eine Teilmenge der zu verfolgenden Spalten angeben, sodass Änderungen an anderen Spalten die aktuelle Version direkt aktualisieren, anstatt einen neuen Verlaufsdatensatz zu erzeugen.

Im folgenden Beispiel wird die city Spalte aus der Verlaufsverfolgung ausgeschlossen:

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Da city Änderungen nicht nachverfolgt werden, überschreiben Stadtupdates die aktuelle Zeile, anstatt eine neue Version zu erstellen. Die Zieltabelle enthält die folgenden Datensätze:

userId name city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 NULL
125 Mercedes Guadalajara 2 NULL
126 Lilie Cancun 2 NULL

Beispiele für AUTO CDC FROM SNAPSHOT

In den folgenden Abschnitten finden Sie Beispiele dafür, wie AUTO CDC FROM SNAPSHOT verwendet wird, um Momentaufnahmen in SCD-Zieltabellen vom Typ 1 oder Typ 2 zu verarbeiten. Hintergrundinformationen zur Verwendung dieser API finden Sie unter Change Data Capture und Snapshots.

Die Beispiele in diesem Abschnitt umfassen Python- und SQL-Implementierungen. Für die vollständige SQL-Syntax siehe CREATE FLOW (pipelines).

Beispiel: Periodische Schnappschüsse verarbeiten

Verwenden Sie diesen Ansatz, wenn die Schnappschüsse regelmäßig und in der richtigen Reihenfolge eintreffen. Die Python-Schnittstelle verwendet die Pipeline-Update-Reihenfolge für die Versionierung. Für SQL verwendet das Beispiel eine separate Versionstabelle, um zu erkennen, wann ein neuer Snapshot verfügbar ist.

Sie können Momentaufnahmen aus mehreren Quelltypen lesen, einschließlich Delta-Tabellen, Cloudspeicherdateien und JDBC-Verbindungen.

Schritt 1: Erstellen von Beispieldaten

Erstellen Sie eine Tabelle mit Momentaufnahmedaten. Führen Sie den folgenden Code aus einem Notizbuch oder Databricks SQL im explorations Ordner Ihrer Pipeline aus:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Schritt 2: Ausführen von AUTO CDC VOM SNAPSHOT

Du brauchst eine Pipeline, um den Code in diesem Schritt auszuführen. Siehe Entwickeln und Debuggen von ETL-Pipelines mit dem Lakeflow Pipelines-Editor.

Für Python wählen Sie einen Quelltyp für die Snapshot-Ansicht (der Beispiel-Erstellungscode erzeugt eine Delta-Tabelle):

Option A: Aus einer Delta-Tabelle lesen 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Option B: Aus Cloudspeicher lesen 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Option C: Lesen Sie aus JDBC (nur klassische Verarbeitung) 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Füge dann die Zieltabelle und den Flow hinzu. Die SQL-Implementierung verwendet die snapshot_version Tabelle, um zu erkennen, wann ein neuer Snapshot verfügbar ist:

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Nach der ersten Pipelineausführung werden alle Datensätze als aktive Zeilen eingefügt:

userId city __START_AT __END_AT
1 Oaxaca 0 NULL
2 Monterrey 0 NULL
3 Tijuana 0 NULL

Hinweis

Um stattdessen SCD Type 1 zu verwenden und nur den aktuellen Zustand beizubehalten, setzen Sie stored_as_scd_type=1 in Python oder STORED AS SCD TYPE 1 in SQL. In diesem Fall enthält die Zieltabelle die Spalten __START_AT und __END_AT nicht.

Schritt 3: Simulieren einer neuen Momentaufnahme und erneutes Ausführen

Aktualisieren Sie die Quelltabelle, um das Eintreffen eines neuen Snapshots zu simulieren (führen Sie diesen Code aus einem Notizbuch oder einer SQL-Datei im explorations Ordner Ihrer Pipeline aus):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Erneutes Ausführen der Pipeline AUTO CDC FROM SNAPSHOT Vergleicht die neue Momentaufnahme mit dem vorherigen und erkennt, dass Benutzer 1 gelöscht wurden, Die Benutzer 2 und 3 wurden aktualisiert, und Benutzer 4 und 6 wurden eingefügt. Dadurch wird ein Änderungsfeed generiert, und AUTO CDC wird zur Erstellung der Ausgabetabelle verwendet.

Nach der zweiten Ausführung mit SCD Type 2 enthält die Zieltabelle die folgenden Datensätze:

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 NULL
3 Tijuana 0 1
3 Los Angeles 1 NULL
4 Death Valley 1 NULL
6 Kings Canyon 1 NULL

Benutzer 1 wurde beendet (gelöscht). Die Benutzer 2 und 3 verfügen jeweils über zwei Versionen, in denen ihre Stadtänderungen angezeigt werden. Benutzer 4 und 6 wurden neu eingefügt.

Nach der zweiten Ausführung mit SCD Type 1 zeigt die Zieltabelle nur den aktuellen Zustand an:

userId city
2 Carmel
3 Los Angeles
4 Death Valley
6 Kings Canyon

Beispiel: Verarbeiten von Momentaufnahmen mithilfe von Versionsfunktionen

Verwenden Sie diesen Ansatz, wenn Sie explizite Kontrolle über die Reihenfolge von Momentaufnahmen benötigen. Verwenden Sie z. B. diesen Ansatz, wenn mehrere Snapshots gleichzeitig eingehen oder Snapshots in der falschen Reihenfolge eintreffen. Du definierst, wie du den nächsten Snapshot und dessen Versionsnummer auswählst. Die API verarbeitet Momentaufnahmen in aufsteigender Versionsreihenfolge:

  • Wenn sich mehrere Snapshots im Speicher befinden, verarbeitet der Beispielselektor verfügbare Snapshots in Versionsreihenfolge.
  • Wenn ein Snapshot nicht in der Reihenfolge eintrifft (zum Beispiel snapshot_3 nach snapshot_4), schließt die Beispielauswahl ihn aus, weil seine Version nicht größer ist als die zuletzt committierte Version.
  • Wenn keine neuen Snapshots vorliegen, liefert die Versionsfunktion oder Abfrage kein Ergebnis und es findet keine Verarbeitung statt.

Schritt 1: Schnappschussdateien vorbereiten

Erstellen Sie CSV-Dateien, die Snapshotdaten enthalten, und fügen Sie sie einem Volume- oder Cloudspeicherort hinzu. Benennen Sie die Dateien chronologisch (z. B snapshot_1.csv. , snapshot_2.csv).

Jede Datei sollte Spalten für userId und city enthalten. Beispiel:

snapshot_1.csv:

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId city
2 Carmel
3 Los Angeles
4 Death Valley

Schritt 2: Ausführen von AUTO CDC FROM SNAPSHOT mit einer Versionsfunktion

Erstelle in einer Pipeline eine neue Python- oder SQL-Datei im transformations Ordner und füge folgenden Code hinzu. Führen Sie dann die Pipeline aus. Siehe Entwickeln und Debuggen von ETL-Pipelines mit dem Lakeflow Pipelines-Editor.

Python

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Hinweis

Um stattdessen SCD Type 1 zu verwenden, setze stored_as_scd_type=1 in Python oder STORED AS SCD TYPE 1 in SQL.

Nach der Verarbeitung snapshot_1.csventhält die Zieltabelle die folgenden Datensätze:

userId city __START_AT __END_AT
1 Oaxaca 1 NULL
2 Monterrey 1 NULL
3 Tijuana 1 NULL

Nach der Verarbeitung snapshot_2.csventhält die Zieltabelle die folgenden Datensätze:

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 NULL
3 Tijuana 1 2
3 Los Angeles 2 NULL
4 Death Valley 2 NULL

Hinweis

Denken Sie daran, dass die Tabelle für SCD-Typ 1 genau wie die letzte Momentaufnahme aussieht. Der Unterschied besteht darin, dass nachgeschaltete Abfragen den Änderungsfeed verwenden können, um nur geänderte Datensätze zu verarbeiten.

Schritt 3: Hinzufügen neuer Momentaufnahmen

Fügen Sie dem Speicherort eine neue CSV-Datei mit geänderten Daten hinzu (z. B. geänderte Ortswerte, neue Zeilen oder entfernte Zeilen). Starten Sie anschließend die Pipeline erneut, um die neue Momentaufnahme zu bearbeiten.

Einschränkungen

Weitere Ressourcen

  • Zuletzt aktualisiert um