Postgres-wijzigingen opslaan in het lakehouse

Note

De functie Gegevensfeed voor Lakebase-wijzigingen bevindt zich in openbare preview.

Stel Lakebase Change Data Feed (CDF) in op een Postgres-tabel en bekijk vervolgens hoe wijzigingen op rijniveau worden weergegeven in de delta-doeltabel.

Stappen: (1) Schakel wijzigingsopname in → (2) Start de feed → (3) Volg een rij in het lakehouse → (4) Wijzig de rij, zie hoe deze doorloopt

Note

Dit is een snelstartgids. Zie Lakebase Change Data Feed voor volledige documentatie.

Voordat u begint

  • Zorg ervoor dat u Een Postgres-database ophalen hebt voltooid. U hebt een Lakebase-project nodig met de playing_with_lakebase voorbeeldtabel.
  • Een Unity Catalog-catalogus en -schema waarop u de machtiging CREATE TABLE hebt.

Stap 1: Wijzigingsopname inschakelen

Postgres heeft volledige rijgegevens nodig in het write-ahead-logboek om CDF te laten werken. Als u de replica-identiteit op full instelt, geeft u Postgres de opdracht om bij elke wijziging zowel de oude als de nieuwe toestand van de rij vast te leggen.

Voer in de Lakebase SQL-editor het volgende uit:

ALTER TABLE playing_with_lakebase REPLICA IDENTITY FULL;

Meer informatie: Replica-identiteit instellen voor alle tabellen in een schema en deze automatisch toepassen op nieuwe tabellen

Stap 2: De feed starten

Lakebase CDF is geconfigureerd op schemaniveau. Elke huidige en toekomstige tabel in het bronschema wordt automatisch opgenomen, zodat u geen afzonderlijke tabellen kiest.

Open vanuit uw productiebranch het tabblad Gegevensfeed wijzigen en klik op Start. Kies public als bronschema en kies vervolgens een doelcatalogus en -schema voor Unity Catalog. De eerste momentopname begint onmiddellijk en lb_playing_with_lakebase_history wordt weergegeven als een Delta-tabel in uw bestemming.

Dialoogvenster starten met bron- en doelselectie.

Meer informatie: De wijzigingengegevensfeed starten

Stap 3: Volg een rij in het lakehouse

Kies een rij uit Lakebase. Bekijk de rij id=2:

SELECT * FROM playing_with_lakebase WHERE id = 2;

Zoek nu dezelfde rij in de Delta-geschiedenistabel. Schakel over naar een Databricks SQL Warehouse of notebook en voer het volgende uit:

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

Vervang <catalog> en <schema> door de bestemming die u in stap 2 hebt gekozen. U ziet rij id=2 met hetzelfde name en value in Lakebase, plus extra kolommen. De eerste momentopname schreef elke bestaande rij in Delta als een insert gebeurtenis. Dit is wat die rij vertegenwoordigt.

In deze extra kolommen wordt beschreven welk soort gebeurtenis elke rij vertegenwoordigt (_pg_change_type), wanneer deze is opgetreden (_timestamp) en de Postgres-bestelgegevens (_pg_lsn, _pg_xid).

Meer informatie: Schema van de doeltabel | Toewijzing van gegevenstypen

Stap 4: De rij wijzigen, zien hoe deze doorloopt

Terug in de Lakebase SQL-editor, werk rij id=2 bij:

UPDATE playing_with_lakebase SET value = 55.5 WHERE id = 2;

Wacht enkele seconden totdat de wijziging wordt weergegeven in de feed en voer vervolgens een nieuwe query uit op de geschiedenistabel:

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

Delta-geschiedenistabel met drie rijen voor id=2: update_preimage, update_postimage en invoegen

Rij id=2 wordt nu drie keer weergegeven: het origineel insert, een update_preimage met de oude waarde en een update_postimage met de nieuwe waarde. Elke wijziging aan het record wordt een nieuw item in de geschiedenis, zodat u altijd over een volledig auditspoor beschikt. Verwijderingen werken op dezelfde manier, waarbij één rij wordt toegevoegd met _pg_change_type = 'delete'.

Meer informatie: Veelvoorkomende wijzigingspatronen | Downstream-pijplijnen bouwen

Volgende stappen 

  • Last updated on