Speichern Sie Postgres-Änderungen im Lakehouse

Note

Das Feature "Lakebase Change Data Feed" befindet sich in der öffentlichen Vorschau.

Richten Sie lakebase Change Data Feed (CDF) in einer Postgres-Tabelle ein, und beobachten Sie dann, wie Änderungen auf Zeilenebene in der Zieldelta-Tabelle angezeigt werden.

Schritte: ① Aktivieren Sie die Änderungserfassung → ② Starten Sie den Feed → ③ Verfolgen Sie eine Zeile bis ins Lakehouse → ④ Ändern Sie die Zeile, und beobachten Sie, wie sie weiterfließt

Note

Dies ist eine Schnellstartanleitung. Vollständige Dokumentation finden Sie unter Lakebase Change Data Feed.

Bevor Sie anfangen

  • Stellen Sie sicher, dass Sie Eine Postgres-Datenbank beziehen abgeschlossen haben. Sie benötigen ein Lakebase-Projekt mit der playing_with_lakebase Beispieltabelle.
  • Ein Unity Catalog-Katalog und -Schema, für das bzw. die Sie über CREATE TABLE Berechtigungen verfügen.

Schritt 1: Aktivieren der Änderungserfassung

Postgres benötigt vollständige Zeilendaten im Schreib-Ahead-Protokoll, damit CDF funktioniert. Das Setzen der Replikatidentität auf „FULL“ bewirkt, dass Postgres bei jeder Änderung sowohl den alten als auch den neuen Zustand der Zeile aufzeichnet.

Führen Sie im Lakebase SQL-Editor Folgendes aus:

ALTER TABLE playing_with_lakebase REPLICA IDENTITY FULL;

Weitere Informationen: Festlegen der Replikatidentität für alle Tabellen in einem Schema und automatisches Anwenden auf neue Tabellen

Schritt 2: Starten des Feeds

Lakebase CDF ist auf Schemaebene konfiguriert. Jede aktuelle und zukünftige Tabelle im Quellschema wird automatisch eingeschlossen, sodass Sie keine einzelnen Tabellen auswählen.

Öffnen Sie in Ihrer Produktionszweigung die Registerkarte " Datenfeed ändern ", und klicken Sie auf "Start". Wählen Sie public als Quellschema und dann einen Unity-Zielkatalog und ein Zielschema aus. Die erste Momentaufnahme beginnt sofort und lb_playing_with_lakebase_history wird als Delta-Tabelle in Ihrem Ziel angezeigt.

Dialogfeld

Weitere Informationen: Starten des Änderungsdatenfeeds

Schritt 3: Verfolgen Sie eine Zeile bis in das Lakehouse

Wählen Sie eine Zeile aus Lakebase aus. Sehen Sie sich die Zeile id=2an:

SELECT * FROM playing_with_lakebase WHERE id = 2;

Suchen Sie nun dieselbe Zeile in der Delta-Verlaufstabelle. Wechseln Sie zu einem Databricks SQL Warehouse oder Notizbuch, und führen Sie Folgendes aus:

SELECT * FROM <catalog>.<schema>.lb_playing_with_lakebase_history
WHERE id = 2;

Ersetzen Sie <catalog> und <schema> durch das Ziel, das Sie in Schritt 2 ausgewählt haben. Sie sehen die Zeile id=2 mit demselben name und value wie in Lakebase sowie zusätzlichen Spalten. Der initiale Snapshot schrieb jede vorhandene Zeile als insert-Ereignis in Delta, was von dieser Zeile repräsentiert wird.

Diese zusätzlichen Spalten beschreiben, welche Art von Ereignis jede Zeile darstellt (_pg_change_type), wann es passiert ist (_timestamp) und welche Informationen zur Postgres-Reihenfolge (_pg_lsn, _pg_xid).

Weitere Informationen: Schema der Zieltabelle | Datentypzuordnung

Schritt 4: Zeile ändern und den Durchlauf verfolgen

Zurück im Lakebase SQL Editor, Zeile id=2aktualisieren:

UPDATE playing_with_lakebase SET value = 55.5 WHERE id = 2;

Warten Sie einige Sekunden, bis die Änderung im Feed angezeigt wird, und fragen Sie dann die Verlaufstabelle erneut ab:

SELECT id, value, _pg_change_type, _timestamp
FROM <catalog>.<schema>.lb_playing_with_lakebase_history
WHERE id = 2
ORDER BY _pg_lsn DESC;

Delta-Verlaufstabelle mit drei Zeilen für id=2: update_preimage, update_postimage und insert

Zeile id=2 wird jetzt dreimal angezeigt: das Original insert, ein update_preimage mit dem alten Wert und ein update_postimage mit dem neuen Wert. Jede Änderung an der Zeile wird zu einer neuen Verlaufszeile, sodass Sie immer über einen vollständigen Überwachungspfad verfügen. Löschvorgänge funktionieren auf die gleiche Weise, wobei eine Zeile mit _pg_change_type = 'delete' angehängt wird.

Weitere Informationen: Allgemeine Änderungsmuster | Erstellen nachgeschalteter Pipelines

Nächste Schritte

  • Last updated on