Lakebase-gegevensfeed wijzigen

Note

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

Wat is de wijzigingsgegevensfeed van Lakebase?

Lakebase introduceert een systeemeigen CDF (Change Data Feed), het ontgrendelen van uw operationele gegevens voor downstream-pijplijnen, modellen en toepassingen. Elke invoeg-, update- en verwijderbewerking op een Lakebase Postgres-tabel wordt vastgelegd uit het write-ahead-logboek en wordt opgeslagen als een nieuwe rij in een door Unity Catalog beheerde Delta-tabel, elke ~15 seconden gebatcheerd en leeggemaakt. De wijzigingsgeschiedenis wordt opgeslagen in een open indeling die elke berekeningsengine kan lezen.

De doeltabellen hebben dezelfde structuur als Delta Change Data Feed: elke rij bevat een _pg_change_type, een LSN, een transactie-id en een tijdstempel. Operationele wijzigingen worden een volwaardige bron voor ETL-, audit- en downstreamafnemers, zonder een externe CDC-stack op te zetten.

Lakebase CDF-gegevensstromen van Postgres via wal2delta naar Delta-tabellen in Unity Catalog.

Gebruikssituaties

Lakebase CDF brengt operationele gegevens naar het lakehouse, zodat downstream-pijplijnen en toepassingen kunnen reageren op wijzigingen wanneer ze plaatsvinden.

Gebruiksituatie Description
ETL-pijplijnen Gebruik Lakebase als bronzen bron voor medallion-pipelines. Bouw incrementele Lakeflow-pijplijnen of Spark Structured Streaming-taken op basis van de wijzigingenfeed en werk downstream zilver- en goudtabellen bij.
Auditlogboeken Onderhoud een volledige, doorzoekbare geschiedenis van elke invoeging, update en verwijdering op een Lakebase-tabel voor naleving en forensische gegevens. De geschiedenis is onveranderbare Delta.
Externe systemen Sla Lakebase-wijzigingsgegevens op in een open formaat dat elk systeem kan gebruiken. Omdat de bestemming een Delta-tabel is in Unity Catalog, hebben externe systemen en niet-Databricks-lezers rechtstreeks toegang tot de feed.

Deze preview inschakelen

Een werkruimtebeheerder moet de previewfunctie Lakebase Change Data Feed inschakelen op de pagina Previews van de werkruimte.

Requirements

  • Automatisch schalen van Lakebase: Een Lakebase Autoscaling-project met Postgres 17.
  • Brondatabase: Tabellen moeten zich in de databricks_postgres database in Lakebase bevinden. Elk project wordt gemaakt met deze standaarddatabase. Dit is een bekende beperking.
  • Unity Catalog: De identiteit die CDF configureert, heeft USE CATALOG, USE SCHEMA en CREATE TABLE nodig op de doelcatalogus en het doelschema. Zie Machtigingen verlenen voor een object.
  • Standaardopslag: Doelcatalogussen die zijn geconfigureerd met standaardopslag, worden niet ondersteund.
  • Lakebase-project: Uw Postgres-rol vereist CAN MANAGE-machtigingen voor het Lakebase-project. Project eigenaren hebben standaard CAN MANAGE. Zie Projectmachtigingen beheren.
  • Gegevenstypen: Zie Toewijzing van gegevenstypen. Typen zonder directe Delta-equivalent worden opgeslagen als STRING.

Lakebase CDF instellen

Om te beginnen stelt u replica identity full in voor de tabellen die u in de feed wilt opnemen (stap 1) en start u CDF in de Lakebase-app (stap 2). Uw gegevens worden weergegeven als Delta-tabellen lb_<table_name>_history in de Unity Catalog-catalogus en het schema dat u kiest.

Stap 1: Replica-identiteit instellen

Een Lakebase-tabel moet REPLICA IDENTITY FULL ingesteld hebben om aan CDF deel te nemen. Postgres registreert standaard alleen de primaire sleutel wanneer een rij wordt bijgewerkt of verwijderd. Door full replica identity in te stellen, zorgt u ervoor dat Postgres zowel de toestand van de rij vóór als na de wijziging vastlegt in de write-ahead log, die CDF nodig heeft om een volledige wijzigingsgeschiedenis op te bouwen.

U kunt deze opdrachten uitvoeren in de Lakebase SQL Editor of een Postgres-client.

Eén tabel

ALTER TABLE <table_name> REPLICA IDENTITY FULL;

Alle bestaande tabellen in een schema

Als u een replica-identiteit wilt instellen voor elke bestaande tabel in een schema (public in dit voorbeeld), voert u het volgende uit:

DO $$
DECLARE r record;
BEGIN
  FOR r IN
    SELECT table_schema, table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
      AND table_type = 'BASE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %I.%I REPLICA IDENTITY FULL;',
      r.table_schema, r.table_name
    );
  END LOOP;
END $$;

Automatisch toepassen op toekomstige tabellen

Als u wilt dat elke zojuist gemaakte tabel automatisch wordt ontvangen REPLICA IDENTITY FULL, installeert u een Postgres-gebeurtenistrigger. Deze wordt na elke CREATE TABLE bewerking uitgevoerd en stelt de identiteit in de nieuwe tabel in:

CREATE OR REPLACE FUNCTION public.set_full_replica_identity()
RETURNS event_trigger
LANGUAGE plpgsql
AS $$
DECLARE
  obj record;
BEGIN
  FOR obj IN
    SELECT * FROM pg_event_trigger_ddl_commands()
    WHERE command_tag = 'CREATE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %s REPLICA IDENTITY FULL;',
      obj.object_identity
    );
  END LOOP;
END $$;

CREATE EVENT TRIGGER set_full_replica_identity_on_create
ON ddl_command_end
WHEN TAG IN ('CREATE TABLE')
EXECUTE FUNCTION public.set_full_replica_identity();

Combineer de gebeurtenistrigger met de lus op het vorige tabblad om zowel bestaande als toekomstige tabellen in één installatie te behandelen.

Controleer voor welke tabellen replica identity is ingesteld

Als u wilt zien welke tabellen in een schema een replica-identiteit hebben geconfigureerd, voert u het volgende uit:

SELECT n.nspname AS table_schema,
       c.relname AS table_name,
       CASE c.relreplident
         WHEN 'd' THEN 'default'
         WHEN 'n' THEN 'nothing'
         WHEN 'f' THEN 'full'
         WHEN 'i' THEN 'index'
       END AS replica_identity
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind = 'r'
  AND n.nspname = 'public'
ORDER BY n.nspname, c.relname;

Alleen rijen met replica_identity = 'full' zijn gereed voor CDF.

Stap 2: De wijzigingengegevensfeed starten

Lakebase CDF is geconfigureerd op schemaniveau. Zodra dit is gestart, wordt elke huidige en toekomstige tabel in het bronschema opgenomen in de feed.

  1. Open in uw Azure Databricks werkruimte Lakebase Postgres vanuit de app-switcher (rechtsboven).
  2. Selecteer uw Lakebase-project en de vertakking die u wilt gebruiken (bijvoorbeeld productie of hoofd).
  3. Open Branch overview door op de branchnaam in de breadcrumb bovenaan te klikken en klik vervolgens op het tabblad Lakebase CDF.
  4. Klik op Start.
  5. In het configuratiedialoogvenster:
    • Database: Standaard ingesteld op databricks_postgres.
    • Schema: Selecteer het bronschema van Postgres.
    • Naar catalogus: Selecteer de doelcatalogus van Unity Catalog.
    • Schema: Selecteer het doel-Unity Catalog-schema.
  6. Klik op Start om de feed te starten.

Overzicht van branches met het tabblad Lakebase CDF, waarop Start en schemaconfiguratie worden weergegeven.

Tabellen worden in het doel weergegeven als lb_<table_name>_history. Als u ze wilt vinden, opent u Catalogus in de zijbalk, gaat u naar de doelcatalogus en het schema en opent u het tabblad Tabellen .

Het CDF-tabblad Lakebase heeft twee subtabbladen:

Subtabbladen geven de toewijzing en voortgang per tabel weer.

  • Schemas: Een lijst met elk bronschema, de doelcatalogus en het schema in Unity Catalog en een status.
  • Tabellen: Geeft een lijst weer van elke brontabel, de doeltabel lb_<table_name>_history , de status (Streaming of Snapshotting), vastgelegde LSN (hoe ver de feed naar Delta is geschreven, weergegeven als - in de eerste momentopname) en laatste update (laatste keer dat de tabel wijzigingen heeft ontvangen).

U kunt ook de feedstatus van Postgres inspecteren door dit uit te voeren in de Lakebase SQL-editor:

SELECT * FROM wal2delta.tables;

Het resultaat bevat table_oid, status (STREAMING of SNAPSHOTTING), committed_lsnen last_write_time per tabel.

Important

Wat is wal2delta? Lakebase CDF wordt mogelijk gemaakt door de wal2delta Postgres-extensie, die wordt uitgevoerd in de Lakebase-rekenkracht. Het maakt gebruik van logische decodering voor het vastleggen van wal-wijzigingen (write-ahead log) en schrijft ze naar Delta-tabellen in Unity Catalog.

Schema van de doeltabel

CDF schrijft één Delta-tabel per brontabel, met de naam lb_<table_name>_history in uw doelcatalogus en -schema. Naast de bronkolommen bevat elke rij deze systeemkolommen:

Rubriek Typ Description
_pg_change_type Tekst Bewerkingstype: insert, delete, update_preimage, of update_postimage.
_pg_lsn BIGINT Postgres-logboekreeksnummer.
_pg_xid INTEGER Postgres-transactie-id.
_timestamp TIMESTAMP Tijdstempel toen de wijziging werd verwerkt (zonder tijdzone).
_sort_by BIGINT Monotone sorteersleutel die wordt gebruikt om alle wijzigingen te orden.

Algemene wijzigingspatronen

  • Eerste momentopname: De eerste keer dat CDF wordt uitgevoerd op een bestaande Lakebase-tabel, wordt elke bestaande rij geschreven met _pg_change_type = 'insert'.
  • Updates: Een update produceert twee rijen: één met _pg_change_type = 'update_preimage' (oude rij) en één met _pg_change_type = 'update_postimage' (nieuwe rij).
  • Verwijderd: Een verwijderbewerking produceert één rij met _pg_change_type = 'delete'.

Dit zijn dezelfde wijzigingsevenementen als Delta Change Data Feed, dus dezelfde downstreampatronen zijn van toepassing.

Operationeel gedrag

  • Naamgevingsconflicten: Als twee brontabellen naar dezelfde doelnaam worden toegewezen (bijvoorbeeld als sales.users en marketing.users beide naar lb_users_history worden toegewezen), schrijft CDF de eerste naar lb_users_history en voorziet het de tweede automatisch van een achtervoegsel als lb_users_history_1. U kunt de naam van beide doeltabel in Unity Catalog wijzigen en de feed blijft werken.
  • Bereik op schemaniveau: Wanneer u CDF start in een Lakebase-schema, wordt elke huidige en toekomstige tabel in dat schema opgenomen. Lege tabellen worden overgeslagen — een tabel moet ten minste één rij bevatten om in het doelbestand te worden weergegeven.
  • Verwijderde brontabellen: Als u een tabel in Lakebase neer zet, blijft de doel-Delta-tabel in Unity Catalog behouden.

Downstream-pijplijnen bouwen

Lakebase CDF is ontworpen voor downstream-pijplijnen die reageren op operationele wijzigingen. In de onderstaande patronen ziet u drie manieren om de feed te gebruiken, gesorteerd van eenvoudigst naar meest flexibel.

Voorbeeldscenario. Een e-commerce-app registreert orders in een Postgres-tabel orders , elke rij met een item_id en quantity. Het logistieke team heeft live voorraadniveaus nodig. Met CDF wordt elke wijziging in orders opgeslagen in de lb_orders_history Delta-tabel in Unity Catalog. Downstream-pipelines lezen die change feed en werken een inventory_levels-tabel bij wanneer een order wordt geplaatst, bewerkt of geannuleerd.

Huidige inventaris berekenen met een gerealiseerde weergave

Het eenvoudigste patroon is een gerealiseerde SQL-weergave over de geschiedenistabel. De MV wordt incrementeel vernieuwd zodra nieuwe wijzigingsgebeurtenissen binnenkomen, en downstreamafnemers bevragen deze net als elke andere tabel.

CREATE MATERIALIZED VIEW inventory_levels AS
SELECT
  item_id,
  SUM(
    CASE
      -- New orders (and the "new half" of updates) decrement inventory
      WHEN _pg_change_type IN ('insert', 'update_postimage') THEN -quantity
      -- Cancellations (and the "old half" of updates) restore inventory
      WHEN _pg_change_type IN ('delete', 'update_preimage') THEN quantity
      ELSE 0
    END
  ) AS current_inventory,
  MAX(_timestamp) AS last_transaction_ts,
  MAX(_pg_lsn) AS last_lsn
FROM lb_orders_history
GROUP BY item_id;

De twee rijen die voor elke update worden geproduceerd, annuleren elkaar, met uitzondering van de nettowijziging, zodat de lopende som correct blijft terwijl orders worden bewerkt.

Wijzigingen streamen met declaratieve Spark-pijplijnen

Gebruik Lakeflow-pijplijnen om voor een gestructureerde medaillonarchitectuur bronzen, zilveren en goudtabellen te definiëren. Lakeflow-pijplijnen voeren deze uit als één verbonden pijplijn, waarbij controlepunten en afhankelijkheidsbeheer voor u worden beheerd.

import dlt
from pyspark.sql import functions as F

@dlt.table
def inventory_adjustments():
    return (
        spark.readStream.table("<catalog>.<schema>.lb_orders_history")
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .select("item_id", "delta", "_timestamp")
    )

@dlt.expect_or_drop("non_negative_stock", "on_hand >= 0")
@dlt.table
def inventory_levels():
    return (
        spark.read.table("LIVE.inventory_adjustments")
        .groupBy("item_id")
        .agg(F.sum("delta").alias("on_hand"))
    )

inventory_adjustments leest lb_orders_history incrementeel met readStream en genereert een delta per gebeurtenis. inventory_levels aggregeert op basis van item_id om de huidige voorraad te berekenen. De verwachting is dat rijen die de voorraad negatief zouden maken, worden weggelaten, wat wijst op een bug upstream.

Zie Zelfstudie: Een ETL-pijplijn bouwen met behulp van change data capture voor een volledige end-to-end-handleiding.

Aangepaste verwerking met Spark Structured Streaming

Wanneer u volledige controle nodig hebt, bijvoorbeeld aangepaste samenvoegingen, neveneffecten of meerdere sinks, leest u de geschiedenistabel rechtstreeks met Spark Structured Streaming en gebruikt foreachBatch u deze om naar uw bestemming te schrijven.

from pyspark.sql import functions as F
from delta.tables import DeltaTable

def update_inventory(batch_df, batch_id):
    deltas = (
        batch_df
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .groupBy("item_id")
        .agg(F.sum("delta").alias("delta"))
    )

    target = DeltaTable.forName(spark, "<catalog>.<schema>.inventory_levels")
    (target.alias("t")
        .merge(deltas.alias("s"), "t.item_id = s.item_id")
        .whenMatchedUpdate(set={"on_hand": F.expr("t.on_hand + s.delta")})
        .whenNotMatchedInsert(values={"item_id": "s.item_id", "on_hand": "s.delta"})
        .execute())

(spark.readStream.table("<catalog>.<schema>.lb_orders_history")
    .writeStream
    .foreachBatch(update_inventory)
    .option("checkpointLocation", "/Volumes/<catalog>/<schema>/checkpoints/inventory_levels")
    .start())

Elke microbatch aggregert de wijzigingsgebeurtenissen door item_id en voegt de netto-delta's samen in inventory_levels.

Incrementeel per ontwerp. Elke lb_<table_name>_history tabel is een Delta-tabel waaraan alleen kan worden toegevoegd. Elke bronwijziging wordt vastgelegd als een nieuwe rij, waarbij _pg_change_type de bewerking markeert. Databricks SQL gerealiseerde weergaven, Lakeflow-pijplijnenstromen en Spark Structured Streaming-taken verwerken alle nieuwe rijen incrementeel vanuit het Delta-transactielogboek, dus downstreampijplijnen werken alleen evenredig met wat is gewijzigd. U hoeft deltawijzigingsgegevensfeed niet in te schakelen in de geschiedenistabel omdat wijzigingssemantiek al in de rijgegevens is gecodeerd.

Koppeling van gegevenstypen

CDF ondersteunt de meeste standaard PostgreSQL primitieve typen. Typen zonder directe Delta-equivalent worden opgeslagen als STRING.

PostgreSQL-type Azure Databricks Delta-type Opmerkingen
BOOLEAN BOOLEAN
INT, SMALLINT, BIGINT INT, SMALLINT, BIGINT
TEKST, VARCHAR, CHAR STRING
JSONB STRING Opgeslagen als een JSON-tekenreeks.
ENUM STRING Opgeslagen als het enum-label.
NUMERIEK / DECIMAAL DECIMAAL OF TEKENREEKS Gebruikt indien mogelijk de precisie/schaal van de bron. Voert verliesloze schaalaanpassing uit voor incompatibele precisie-/schaalwaarden. Valt terug op STRING wanneer de precisie hoger is dan 38 of wanneer precisie/schaal niet zijn gedefinieerd (onbegrensde NUMERIC). Alle kolommen van het type NUMERIC/DECIMAL kunnen NULL-waarden bevatten, omdat NaN-waarden worden toegewezen aan NULL. Zie Numerieke postgreSQL-typen.
DATE DATE
TIMESTAMP TIMESTAMP_NTZ
TIMESTAMPTZ TIMESTAMP
FLOAT, DOUBLE FLOAT, DOUBLE

Typen die zijn opgeslagen als TEKENREEKS:

  • Geografie/geometrie (PostGIS): Typen uit de PostGIS-extensie (bijvoorbeeld geometry, geography).
  • Vector (pgvector): Het vector type van de pgvector-extensie.
  • Samengestelde/structtypen: Aangepaste typen gedefinieerd met CREATE TYPE ... AS (field_name type, ...). Dit zijn rijachtige typen met benoemde velden.
  • Map: Sleutel-waardetypen van het type Map, zoals hstore (van de hstore-extensie). Postgres heeft geen ingebouwd kaarttype. hstore is de gebruikelijke manier om sleutel-waardeparen op te slaan in een kolom.

Schemawijzigingen beheren

  • Als u de naam van een tabel wijzigt in Postgres (bijvoorbeeld ALTER TABLE users RENAME TO customers) kan de feed worden voortgezet. De naam van de doel-Delta-tabel verandert niet — deze blijft lb_users_history.
  • Schemawijzigingen (het toevoegen van een kolom, het verwijderen van een kolom of het wijzigen van het gegevenstype van een kolom) activeren een nieuwe momentopname van de betreffende tabel. CDF leest de hele tabel opnieuw uit Postgres en herschrijft deze naar de delta-doeltabel.

Lakebase CDF uitschakelen

Als u CDF uitschakelt, wordt de feed voor alle Lakebase-schema's in het project gestopt.

  1. Open in uw Azure Databricks werkruimte Lakebase Postgres vanuit de app-switcher (rechtsboven).
  2. Selecteer uw Lakebase-project en de vertakking waar u CDF hebt geconfigureerd.
  3. Open Branch overview door op de branchnaam in de breadcrumb bovenaan te klikken en klik vervolgens op het tabblad Lakebase CDF.
  4. Klik op Uitschakelen. Bekijk in het bevestigingsdialoogvenster de waarschuwing dat wijzigingen niet meer naar Delta-tabellen stromen en klik vervolgens nogmaals op Uitschakelen om te bevestigen.

Als u CDF uitschakelt, wordt uw rekenproces niet opnieuw opgestart.

Beperkingen en probleemoplossing

U kunt de status van elke tabel bekijken (snapshot maken, overgeslagen of streamen) op het tabblad Lakebase CDF, of door dit in Lakebase uit te voeren:

SELECT * FROM wal2delta.tables;

Veelvoorkomende redenen waarom een tabel niet wordt weergegeven in de feed:

  • REPLICA IDENTITY FULL niet ingesteld: Voer ALTER TABLE <table_name> REPLICA IDENTITY FULL; uit voor de tabel. Zie stap 1: De replica-identiteit volledig instellen.
  • Gepartitioneerde tabellen: Gepartitioneerde tabellen in Lakebase worden niet ondersteund. Een schema dat gepartitioneerde tabellen bevat, zorgt ervoor dat deze tabellen mislukken.
  • Lege tabellen: Een tabel met nul rijen wordt overgeslagen tot ten minste één rij bestaat.
  • Privé-eindpunt voor doelopslag: Lakebase CDF wordt niet ondersteund wanneer de beheerde opslag voor uw doelcatalogus van Unity Alleen toegankelijk is via een privé-eindpunt (bijvoorbeeld wanneer openbare netwerktoegang tot het opslagaccount is uitgeschakeld). Configureer als tijdelijke oplossing een catalogus waarvan de beheerde opslag openbaar bereikbaar is en gebruik die catalogus als uw CDF-bestemming.

Volgende stappen 

  • Last updated on