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".

• Um den Change Data Feed zu lesen, müssen Sie Databricks Runtime 18 LTS oder höher verwenden, auf Classic Compute, serverless Compute oder Databricks SQL.

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

• Die materialisierte Ansicht muss Zeilenverfolgung aktiviert haben. 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');
  

• Die materialisierten View-Metadaten müssen so synchronisiert werden, dass sie außerhalb ihrer eigenen Pipeline lesbar sind:

  • Für eine materialisierte Ansicht, die in einer Pipeline erstellt wurde, setzen pipelines.externalMetadata.enabled Sie in der Pipeline-Konfiguration:

    {
      "configuration": {
        "pipelines.externalMetadata.enabled": "true"
      }
    }
    

  • Für eine eigenständige materialisierte Ansicht führen Sie den folgenden Befehl einmal auf jeder materialisierten Ansicht aus. Dieser Befehl benötigt die externe Zugriffsvorschau für materialisierte Ansichten und Streaming-Tabellen. Siehe Aktivieren des externen Datenzugriffs für Streamingtabellen und materialisierte Ansichten.

    REPAIR TABLE my_mv SYNC METADATA;
    

Man liest den Änderungsdaten-Feed aus einer materialisierten Ansicht genauso wie aus anderen Delta-Tabellen, indem man die Funktion table_changes() , eine Streaming-Lektüre oder die Option readChangeFeed verwendet. Für Syntax und Beispiele in SQL und Python siehe Use change data feed in Azure Databricks.

Um einen materialisierten View Change Data Feed aus einer Databricks SQL-Materialisier-Ansicht oder Streaming-Tabelle zu lesen, muss diese materialisierte Ansicht oder Streaming-Tabelle ebenfalls den Kanal PREVIEW verwenden:

CREATE OR REFRESH STREAMING TABLE sales
TBLPROPERTIES ('pipelines.channel' = 'preview')
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

Einschränkungen

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 sie 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 zwischen einer Spalte unterschieden werden, AUTO CDC die nicht in einem Änderungsdatensatz vorhanden ist, wodurch der Zielwert unverändert bleiben soll, und einer Spalte, die explizit festgelegt nullist, die den Zielwert nullmit überschreiben soll. Behandelt standardmäßig IGNORE NULL UPDATES jede null Markierung als "Nicht aktualisieren"-Markierung, sodass keine explizite Anwendung möglich nullist. Um diese Mehrdeutigkeit zu beheben, wählen Sie eine der folgenden drei Methoden aus:

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

Wählen Sie als Faustregel aus COLUMNS TO UPDATE , wann der Produzent weiß, welche Spalten in jedem Datensatz geändert wurden, und diese Informationen in einer Quellspalte übertragen können, z. B. wenn mehrere Produzenten in dieselbe Quelle schreiben oder die Einrichtung aktualisierbarer Spalten im Laufe der Zeit wächst. Wählen Sie aus IGNORE NULL UPDATES ON , wann der Pipelinebesitzer den festen Satz aktualisierbarer Spalten im Voraus kennt und bevorzugt, sie im Pipelinecode zu steuern.

Im folgenden Beispiel wird eine Quellspalte mit dem Namen columnsToUpdate verwendet, um zu steuern, welche Spalten die einzelnen Datensatzänderungen aktualisieren, einschließlich spalten, die explizit auf null:

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.

Bitemporales 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: Out-of-Order-Update

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.