Delta-Änderungsdatenfeed in der Spiegelung für Fabric

Delta change data feed (CDF) erfasst Einfügungen, Aktualisierungen und Löschungen und wendet sie dann auf Delta Lake-Tabellen in OneLake an. Dieser Fluss unterstützt nahezu Echtzeitanalysen ohne vollständige Neuladungen oder schwere ETL-Pipelines.

• Verarbeitet fortlaufend inkrementelle Änderungen (deltabasiert).
• Verwendet die änderungsbasierte Verarbeitung.
• Unterstützt die nachgeschaltete inkrementelle Verarbeitung.
• Ist für alle Spiegelungsquellen verfügbar, einschließlich offener Spiegelungspartner.

Voraussetzungen

• Eine Kapazität für Microsoft Fabric (F2 oder höher) oder eine Fabric-Testversion.
• Eine gespiegelte Datenbank in einem Fabric-Arbeitsbereich.
• Ein Fabric Lakehouse ist im gleichen oder einem anderen Arbeitsbereich vorhanden (zum Abfragen von CDF-Daten)

Aktivieren des Delta-Änderungsdatenfeeds im Fabric-Portal

Aktivieren Sie den Delta-Änderungsdatenfeed für jede gespiegelte Datenbank.

1. Wählen Sie für eine gespiegelte Quelle das Zahnradsymbol aus, um das Konfigurationsdashboard zu öffnen.
2. Aktivieren Sie unter Delta-Tabellenverwaltung das Kontrollkästchen zum Aktivieren des Delta-Änderungsdatenfeeds.

Aktivieren des Delta-Änderungsdatenfeeds mithilfe von APIs

Informationen zum Aktivieren des Delta-Änderungsdatenfeeds mithilfe von APIs finden Sie unter Aktivieren des Delta-Änderungsdatenfeeds für eine gespiegelte Datenbank.

Delta-Änderungsdatenfeed für vorhandene Tabellen aktivieren

Wenn Sie Ihre gespiegelte Datenbank erstellt haben, bevor CDF verfügbar wurde, können Sie CDF mithilfe der Fabric REST-API aktivieren. Der Prozess ruft die aktuelle Definition ab, fügt die CDF-Eigenschaft hinzu und aktualisiert die Definition.

1. Die gespiegelte Datenbankdefinition abrufen:

   POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/mirroredDatabases/{mirroredDatabaseId}/getDefinition
   

2. Dekodieren Sie die mirroring.json Payload aus der Antwort. Die Nutzlast ist base64-codiert.

3. "enableDeltaChangeDataFeed": true zum Abschnitt target.typeProperties hinzufügen:

   {
     "properties": {
       "source": { ... },
       "target": {
         "type": "MountedRelationalDatabase",
         "typeProperties": {
           "defaultSchema": "dbo",
           "format": "Delta",
           "enableDeltaChangeDataFeed": true,
           "retentionInDays": 1
         }
       }
     }
   }
   

4. Codieren Sie den aktualisierten JSON-Code erneut in Base64, und aktualisieren Sie dann die Definition:

   POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/mirroredDatabases/{mirroredDatabaseId}/updateDefinition
   

Nachdem Sie die Definition aktualisiert haben, rufen Sie sie erneut ab und bestätigen Sie, dass enableDeltaChangeDataFeedtrue ist.

Änderungsdaten abfragen

1. Erstellen Sie eine Lakehouse-Verknüpfung:

   1. Öffnen oder erstellen Sie in Ihrem Fabric-Arbeitsbereich ein Lakehouse.
   2. Wählen Sie im Lakehouse-Explorer die Option "Neue Verknüpfung" aus.
   3. Wählen Sie Microsoft OneLake als Quelle aus.
   4. Wechseln Sie zur gespiegelten Datenbank, und wählen Sie die Tabelle aus, die Sie abfragen möchten.
   5. Wählen Sie "Erstellen" aus, um die Erstellung der Verknüpfung abzuschließen.

2. Öffnen Sie ein Fabric Notizbuch, das an Ihr Lakehouse angefügt ist, und verwenden Sie eine der folgenden Optionen zum Abfragen von Änderungsdaten:

   • Option A: Änderungen ab einer bestimmten Version lesen
   • Option B: Lesen von Änderungen innerhalb eines Zeitstempelbereichs
   • Option C: Sql-Syntax verwenden

Option A: Änderungen ab einer bestimmten Version lesen

Verwenden Sie die Option readChangeFeed mit startingVersion, um alle CDF-Änderungen ab einer bestimmten Version einer Delta-Tabelle zu lesen. Ersetzen Sie <lakehouse_name> und <table_name> durch den Namen Ihres Lakehouse und Ihrer Tabelle.

df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 0) \
    .table("<lakehouse_name>.<table_name>")

df.show()

Option B: Lesen von Änderungen innerhalb eines Zeitstempelbereichs

Verwenden Sie die Option readChangeFeed zusammen mit startingTimestamp und endingTimestamp, um CDF-Änderungen innerhalb eines bestimmten Zeitfensters zu lesen.

df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingTimestamp", "2025-01-01T00:00:00Z") \
    .option("endingTimestamp", "2025-01-02T00:00:00Z") \
    .table("<lakehouse_name>.<table_name>")

df.show()

Option C: Sql-Syntax verwenden

Verwenden Sie die table_changes() SQL-Funktion, um CDF-Daten abzufragen. Dieser Ansatz funktioniert in jedem Spark SQL-Kontext.

df = spark.sql("""
    SELECT * FROM table_changes('<lakehouse_name>.<table_name>', 0)
""")

df.show()

Nachgeschaltete Verbrauchsoptionen

Delta-Änderungsdatenfeed unterstützt mehrere nachgelagerte Verbrauchspfade:

Preisgestaltung

Wichtige Preisklärungen:

• Die Abrechnung basiert auf der Nutzung. Sie werden nur für die inkrementelle Berechnung in Rechnung gestellt, die verwendet wird, wenn CDF echte Änderungen verarbeitet. Es fallen keine Gebühren für Leerlaufzeiten oder leere Durchläufe an (Zeiträume, in denen keine Änderungen an den Quelldaten erfolgen).
• Core-Mirroring bleibt kostenlos. Das Aktivieren von CDF ändert die Preise für die Kernspiegelung nicht. Kontinuierliche Replikation, Delta Lake-Konvertierung, OneLake-Integration und SQL-Analyseendpunkte bleiben kostenlos.
• Speicher für Spiegelung ist kostenlos. Der Speicher für gespiegelte Daten in OneLake wird nicht separat in Rechnung gestellt. Die Aktivierung von CDF erhöht jedoch den Speicherverbrauch aufgrund zusätzlicher _change_data Dateien.
• CDF ist ein Add-On, kein Ersatz. Die CDF-Abrechnung ist nur für die erweiterte Rechenkapazität vorgesehen. Für Core-Mirroring-Aktivitäten werden keine Gebühren rückwirkend berechnet.
• Sie können Die Kosten steuern, indem Sie CDF selektiv aktivieren. CDF ist auf gespiegelter Datenbankebene aktiviert. Aktivieren Sie CDF nur für die gespiegelten Datenbanken, die eine inkrementelle Verarbeitung benötigen, und belassen Sie die anderen bei der Core-Spiegelung.

Für das vollständige Preismodell, Details zur Verbrauchserfassung und den Abrechnungsumfang siehe Abrechnung für erweiterte Funktionen beim Mirroring.

Verwandte Inhalte

• Erweiterte Funktionen in der Spiegelung für Fabric
• Spiegeln von Ansichten in Fabric
• Abrechnung für erweiterte Funktionen in der Spiegelung
• Microsoft Fabric-Spiegelung für die öffentliche REST-API