Erweiterte AUTO CDC-Themen

Über die grundlegenden AUTO CDC und AUTO CDC FROM SNAPSHOT APIs hinaus können Sie DML für Zieltabellen ausführen, Änderungsdatenfeeds von CDC-Zielen lesen, Verarbeitungsmetriken überwachen, Teilaktualisierungen anwenden und Änderungen mit bitemporaler Speicher nachverfolgen. Eine Einführung in die AUTO CDC APIs finden Sie unter "Auto CDC-APIs: Vereinfachen der Änderungsdatenerfassung mit Pipelines".

Hinzufügen, Ändern oder Löschen von Daten in einer Zielstreamingtabelle

Wenn Ihre Pipeline Tabellen im Unity-Katalog veröffentlicht, können Sie DML-Anweisungen (Data Manipulation Language ) verwenden, einschließlich Einfüge-, Aktualisierungs-, Lösch- und Zusammenführungsanweisungen, um die von AUTO CDC ... INTO Anweisungen erstellten Zielstreamingtabellen zu ändern.

Da die Pipeline eine angegebene SEQUENCE BY Spalte verwendet und geeignete Sequenzierungswerte an die __START_AT Und __END_AT Spalten der Zieltabelle (für SCD Type 2) verteilt, müssen Sie sicherstellen, dass DML-Anweisungen gültige Werte für diese Spalten verwenden, um die richtige Reihenfolge von Datensätzen beizubehalten. Erfahren Sie , wie AUTO CDC funktioniert.

Weitere Informationen zur Verwendung von DML-Anweisungen mit Streamingtabellen finden Sie unter Hinzufügen, Ändern oder Löschen von Daten in einer Streamingtabelle.

Im folgenden Beispiel wird ein aktiver Datensatz mit einer Startsequenz von 5 eingefügt.

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

Einen Änderungsdaten-Feed aus einer AUTO CDC-Zieltabelle lesen

In Databricks Runtime 15.2 und höher können Sie einen Änderungsdatenfeed aus einer Streamingtabelle lesen, die das Ziel von AUTO CDC- oder AUTO CDC FROM SNAPSHOT-Abfragen ist, auf die gleiche Weise, wie Sie einen Änderungsdatenfeed aus anderen Delta-Tabellen lesen. Folgendes ist erforderlich, um den Änderungsdatenfeed aus einer Ziel-Streamingtabelle zu lesen:

• Die Zielstreamingtabelle muss im Unity-Katalog veröffentlicht werden. Siehe Verwenden des Unity-Katalogs mit Pipelines.
• Um den Änderungsdatenfeed aus der Zielstreamingtabelle zu lesen, müssen Sie Databricks Runtime 15.2 oder höher verwenden. Um den Änderungsdatenfeed in einer anderen Pipeline zu lesen, muss die Pipeline für die Verwendung von Databricks Runtime 15.2 oder höher konfiguriert sein.

Sie lesen den Änderungsdatenfeed aus einer Zielstreamingtabelle, die in einer Lakeflow-Pipeline erstellt wurde, auf die gleiche Weise, wie Sie einen Änderungsdatenfeed aus anderen Delta-Tabellen lesen. Weitere Informationen zur Verwendung der Delta-Änderungsdatenfeedfunktionen, einschließlich Beispiele in Python und SQL, finden Sie unter Verwenden des Änderungsdatenfeeds auf Azure Databricks.

Lesen Sie einen Änderungsdatenfeed aus einer materialisierten Ansicht

Du kannst einen Change Data Feed aus einer materialisierten Ansicht lesen, die in einer Lakeflow-Pipeline oder in Databricks SQL erstellt wurde. Nutzen Sie dies, um materialisierte Ansichtsänderungen auf Ziele außerhalb von Azure Databricks zu replizieren oder um eine Historie materialisierter Ansichtsänderungen für Audits und Berichterstattung zu führen.

Materialisierte Ansichten verwenden den automatischen Änderungsdatenfeed, sodass Sie den Änderungsdaten-Feed selbst nicht aktivieren. Stattdessen aktivieren Sie den Änderungsdaten-Feed für jede materialisierte Ansicht, für die Sie es benötigen, indem Sie die folgenden Anforderungen erfüllen. Siehe "Datenfeed für automatische Änderung".

• Zum Lesen des Change Data Feed müssen Sie Databricks Runtime 18 LTS oder höher mit Classic Compute, Serverless Compute oder Databricks SQL verwenden.

• Die materialisierte Ansicht, die Pipeline, die sie erzeugt, oder die Pipeline, die sie liest, müssen den Kanal PREVIEW verwenden.

• Die materialisierte Ansicht muss für die Zeilenverfolgung aktiviert sein. Materialisierte Ansichten auf serverloser Berechnung haben standardmäßig Zeilenverfolgung aktiviert. Siehe Zeilenverfolgung in Azure Databricks. Um zu überprüfen, ob Zeilenverfolgung in einer materialisierten Ansicht aktiviert ist, führen Sie aus:

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Um den Änderungsdatenfeed aus einer materialisierten Ansicht zu lesen, aktivieren Sie die externe Metadaten-Flagge auf der Pipeline oder in der materialisierten Ansicht. Für Anleitungen siehe Wie man Zugriff auf einen Datensatz aktiviert.

Den Änderungsdaten-Feed aus einer materialisierten Ansicht lesen Sie genauso wie aus anderen Delta-Tabellen, indem Sie die Funktion table_changes(), einen Streaming-Lesevorgang oder die Option readChangeFeed verwenden. Syntax und Beispiele zu SQL und Python finden Sie unter Change Data Feed in Azure Databricks verwenden.

Sie können den Änderungsdatenfeed einer materialisierten Ansicht innerhalb einer materialisierten Databricks SQL-Ansicht oder einer Streaming-Tabelle lesen:

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

Begrenzungen

Zusätzlich zu den automatischen Einschränkungen des Änderungsdatenfeeds gelten Folgendes, wenn Sie einen Änderungsdatenfeed aus einer materialisierten Ansicht lesen:

• Der Änderungsdatenfeed enthält unveränderte Zeilen, wenn die materialisierte Ansicht vollständig umgeschrieben ist, und konsolidiert nicht mehrere Aktualisierungen derselben Zeile zu einem einzigen Ereignis. Um diese herauszufiltern, aggregieren Sie den Änderungsdatenfeed, indem Sie alle Spalten gruppieren, um Einfügungen und Löschungen mit denselben Zeilenwerten zu finden.
• Nur Azure Databricks können den Change Data Feed für eine materialisierte Ansicht abfragen. Externe Kunden von Delta Lake und Iceberg können das nicht.
• Innerhalb von Lakeflow-Pipelines kann man einen materialisierten View-Change-Datenfeed nur aus einer anderen Pipeline lesen, und diese Pipeline muss den Kanal PREVIEW verwenden. Das Lesen des Änderungsdatenfeeds einer materialisierten Ansicht in derselben Pipeline, die diese erstellt, wird nicht unterstützt.
• Man kann keinen Vektorsuchindex aus einer materialisierten Ansicht erstellen.

Daten zu Datensätzen abrufen, die durch eine CDC-Abfrage in Pipelines verarbeitet wurden

Die folgenden Metriken werden von AUTO CDC Abfragen erfasst:

• num_upserted_rows: Die Anzahl der Ausgabezeilen, die während einer Aktualisierung in das Dataset eingefügt wurden.
• num_deleted_rows: Die Anzahl vorhandener Ausgabezeilen, die während einer Aktualisierung aus dem Dataset gelöscht wurden.

Die Metrik „num_output_rows“, die für Nicht-CDC-Flows ausgegeben wird, wird nicht für AUTO CDC-Abfragen erfasst.

Anwenden teilweiser Updates

Wenn eine Quelle nur die geänderten Spalten sendet, muss AUTO CDC zwischen einer Spalte unterscheiden, die in einem Änderungsdatensatz fehlt und den Zielwert daher unverändert lassen sollte, und einer Spalte, die explizit auf null gesetzt wird und den Zielwert mit null überschreiben sollte. Standardmäßig behandelt IGNORE NULL UPDATES jede null-Markierung als „Nicht aktualisieren“-Markierung, daher kann es kein explizites null anwenden. Um diese Mehrdeutigkeit zu beheben, wählen Sie eine der folgenden drei Methoden aus:

COLUMNS TO UPDATE kann nicht mit IGNORE NULL UPDATES kombiniert werden und wird für bitemporale Tabellen nicht unterstützt.

Wählen Sie als Faustregel COLUMNS TO UPDATE, wenn der Produzent weiß, welche Spalten sich in den einzelnen Datensätzen geändert haben, und diese Information in einer Quellspalte mitführen kann, z. B. wenn mehrere Datenproduzenten in dieselbe Quelle schreiben oder die Menge der aktualisierbaren Spalten im Laufe der Zeit wächst. Wählen Sie IGNORE NULL UPDATES ON, wenn der Pipelinebesitzer den festen Satz aktualisierbarer Spalten im Voraus kennt und es vorzieht, diese im Pipeline-Code zu steuern.

Im folgenden Beispiel wird eine Quellspalte namens columnsToUpdate verwendet, um zu steuern, welche Spalten jeder Änderungsdatensatz aktualisiert, einschließlich Spalten, die explizit auf null gesetzt sind:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

Die vollständige Parameterreferenz finden Sie unter AUTO CDC INTO (pipelines) und create_auto_cdc_flow.

Bitemporal AUTO CDC

SCD Typ 1 und Typ 2 sind unitemporal: Sie verfolgen Änderungen über eine einzelne Zeitdimension hinweg. Bitemporal erweitert den SCD Typ 2-Verlauf, um Änderungen über zwei Zeitdimensionen nachzuverfolgen und zwischen zwei Perspektiven zu unterscheiden:

• Geschäftszeit: Wann das Ereignis tatsächlich passiert ist.
• Systemzeit: wenn das System das Ereignis aufgezeichnet oder erfasst hat.

Wie SCD Typ 2 bewahrt bitemporal eine vollständige Historie von Datensätzen. Es fügt eine zweite Zeitachse hinzu, sodass Sie sowohl die angezeigten Daten als auch das, was das System an einem beliebigen Punkt in der Vergangenheit glaubte, rekonstruieren können.

Beispielsweise erfasst ein Hedgefonds Aktiendaten aus einem Quellsystem. Der Aktienkurs von Acme Corp ändert sich am 1. Januar, aber der Fonds übernimmt diese Aktualisierung erst am 5. Januar. Bitemporal AUTO CDC lässt den Fonds zwei verschiedene Fragen beantworten: Was der tatsächliche Aktienkurs von Acme Corp am 1. Januar (Geschäftszeit) war und welcher Preis das System glaubte, als der Fonds am 3. Januar Handelsentscheidungen getroffen hat (Systemzeit). Die Möglichkeit, zwischen diesen Zeitachsen zu unterscheiden, ist nützlich für die Prüfung, regulatorische Berichterstattung und die entscheidungsfindung im Finanzwesen.

Um die bitemporale Verarbeitung zu ermöglichen, legen Sie STORED AS BITEMPORAL (SQL) oder stored_as_scd_type="bitemporal" (Python) fest, verwenden Sie SEQUENCE BY die Geschäftszeitspalte und verwenden Sie SYSTEM SEQUENCE BY sie für die Systemzeitspalte. Die Zieltabelle fügt __SYSTEM_START_AT und __SYSTEM_END_AT Spalten zusammen mit den SCD-Typ 2 __START_AT und __END_AT Spalten hinzu. Ausführliche Informationen zur Syntax finden Sie unter AUTO CDC INTO (Pipelines) oder create_auto_cdc_flow.

Bitemporale AUTO CDC-Beispiele

Im folgenden Beispiel wird eine bitemporale Zieltabelle aus einer kleinen Gruppe synthetischer CDC-Ereignisse erstellt. Die bt Spalte weist die Geschäftszeit auf, und die st Spalte hat die Systemzeit.

Python

from pyspark import pipelines as dp

# Source: synthetic CDC events
dp.create_streaming_table(name="cdc_source")

@dp.append_flow(target="cdc_source", once=True)
def load_cdc_source():
  return spark.createDataFrame(
    [
      (1, "x10", "y10", 10, 100),
      (1, "x20", "y20", 20, 200)
    ],
    schema="id INT, x STRING, y STRING, bt INT, st INT",
  )

# Target: bitemporal table
dp.create_streaming_table(name="target_bitemporal")

dp.create_auto_cdc_flow(
  target = "target_bitemporal",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "bt",
  system_sequence_by = "st",
  stored_as_scd_type = "bitemporal"
)

SQL

-- Source: synthetic CDC events
CREATE OR REFRESH STREAMING TABLE cdc_source_sql;

CREATE FLOW cdc_source_sql AS INSERT INTO ONCE
  cdc_source_sql BY NAME
SELECT * FROM VALUES
  (1, 'x10', 'y10', 10, 100),
  (1, 'x20', 'y20', 20, 200)
  AS t(id, x, y, bt, st);

-- Target: bitemporal table
CREATE OR REFRESH STREAMING TABLE target_bitemporal_sql;

CREATE FLOW target_bitemporal_sql AS AUTO CDC INTO
  target_bitemporal_sql
FROM
  stream(cdc_source_sql)
KEYS
  (id)
SEQUENCE BY
  bt
SYSTEM SEQUENCE BY
  st
STORED AS
  BITEMPORAL;

Die folgende Reihenfolge der Änderungen zeigt, wie eine bitemporale Tabelle einen Einfügevorgang, eine Aktualisierung, ein Out-of-Order-Update und ein Löschvorgang für ein einzelnes Unternehmen aufzeichnet. Die Sequenzierungsspalte generiert die __START_AT Spalten und __END_AT (Geschäftszeit) und die Spalte für die Systemsequenzierung generiert die __SYSTEM_START_AT Spalten und __SYSTEM_END_AT (Systemzeit):

Das System behandelt Ereignisse, die in beliebiger Reihenfolge auf beiden Zeitachsen eingehen. Wenn ein Ereignis mit einer früheren Geschäfts- oder Systemzeit als bereits verarbeitete Ereignisse eintrifft, korrigiert das System den betroffenen Verlauf, anstatt ihn nur am Ende anzufügen.

Änderung 1: Einfügen

Firma A wird am 18.07.2025 um 10:01:00 (Business-Zeit) hinzugefügt, jedoch erst um 10:05:00 (Systemzeit) ins System übernommen.

Eingabe:

Ausgabe:

XFv1 gilt ab 10:01:00 Uhr ohne bekanntes Ende. Das System erfuhr von dieser Tatsache um die Systemzeit 10:05:00, wobei kein Ende bekannt ist.

Änderung 2: Aktualisieren

Unternehmen A wurde am 18.07.2025 um 12:15:43 (Geschäftszeit) aktualisiert, und das System verarbeitet das Ereignis um 12:20:00 (Systemzeit). Das System behält sowohl das, was es glaubte, bevor das Update bekannt war, als auch den korrigierten Geschäftsverlauf nach der Aufnahme des Updates.

Eingabe:

Ausgabe:

Es wurde angenommen, dass XFv1 ab 10:01:00 gültig war, ohne bekanntes Ende der Gültigkeit, und das System ging von 10:05:00 bis 12:20:00 davon aus. XFv1 gilt nach aktuellem Kenntnisstand nur bis 12:15:43; eine korrigierte Historie ist ab Systemzeit 12:20:00 ohne bekanntes Ende gültig. XFv2 ist ab 12:15:43 ohne bekanntes Ende gültig und wurde zur Systemzeit 12:20:00 Uhr gelernt.

Änderung 3: Update in falscher Reihenfolge

Ein Update außerhalb der Reihenfolge trifft ein, das angibt, dass Unternehmen A tatsächlich am 18.07.2025 um 12:05:00 Uhr (Geschäftszeit) aktualisiert wurde, aber erst um 12:25:00 Uhr (Systemzeit) erfasst wird. Wenn ein Update später in der Systemzeit eintrifft, aber mit einer vorherigen Geschäftszeit, korrigiert das System die historische Geschäftszeit und behält sowohl das, was es vor dem Out-of-Order-Update als auch dem korrigierten Verlauf glaubte.

Eingabe:

Ausgabe:

XFv1 wurde von 10:01:00 bis 12:15:43 als gültig angesehen, und diese Annahme ist jetzt in der Systemzeit bis 12:25:00 gültig. Das neue Update korrigiert die geschäftliche Gültigkeit von XFv1 so, dass sie um 12:05:00 Uhr endet, wobei der korrigierte Verlauf ab der Systemzeit 12:25:00 Uhr wirksam ist. XFv3 gilt jetzt als von 12:05:00 bis 12:15:43 gültig, wobei diese Erkenntnis in der Systemzeit ab 12:25:00 gültig ist und kein bekanntes Ende hat.

Änderung 4: Löschen

Firma A wird am 18.07.2025 um 12:30:00 Uhr gelöscht, und das System verarbeitet das Ereignis um 12:30:00 Uhr. Da ein Löschvorgang das Ende des Geschäftsbestands der Entität darstellt, erstellt das System keine Ersetzungszeile. XFv2 wird in zwei Zeilen angezeigt, sodass ein vollständiger Audit-Trail sowohl darüber erhalten bleibt, wann das Unternehmen aufhörte zu existieren, als auch darüber, wann das System von der Löschung erfuhr.

Eingabe:

Ausgabe:

XFv2 war ab 12:15:43 gültig, ohne bekannten Endzeitpunkt, und das System ging von 12:20:00 bis 12:30:00 davon aus. Nachdem die Löschung eingelesen wurde, ist bekannt, dass XFv2 nur bis 12:30:00 gültig ist, wobei ab der Systemzeit 12:30:00 ein korrigierter Verlauf gilt.

Welche Datenobjekte werden für die CDC-Verarbeitung in einer Pipeline verwendet?

Wenn Sie die Zieltabelle im Hive-Metaspeicher deklarieren, werden zwei Datenstrukturen erstellt:

• Eine Ansicht mit dem Namen, der der Zieltabelle zugewiesen ist.
• Eine interne Sicherungstabelle, die von der Pipeline zum Verwalten der CDC-Verarbeitung verwendet wird. Diese Tabelle wird benannt, indem __apply_changes_storage_ an den Zieltabellennamen vorangestellt wird.

Wenn Sie z. B. eine Zieltabelle mit dem Namen dp_cdc_target deklarieren, wird eine Ansicht namens dp_cdc_target und eine Tabelle namens __apply_changes_storage_dp_cdc_target im Metastore angezeigt. Fragen Sie die Sicht ab, um auf die verarbeiteten Daten zuzugreifen. Ändern Sie die zugrunde stehenden Tabellen nicht direkt.