Geavanceerde AUTO CDC-onderwerpen

Naast de basis AUTO CDC - en AUTO CDC FROM SNAPSHOT API's kunt u DML uitvoeren op doeltabellen, wijzigingen van gegevensfeeds van CDC-doelen lezen, metrische verwerkingsgegevens bewaken, gedeeltelijke updates toepassen en wijzigingen bijhouden met bitemporale opslag. Zie AUTO CDC.

Gegevens toevoegen, wijzigen of verwijderen in een doelstreamingtabel

Als uw pijplijn tabellen publiceert naar Unity Catalog, kunt u DML-instructies ( Data Manipulat Language ) gebruiken, zoals invoeg-, update-, verwijder- en samenvoeginstructies om de doelstreamingtabellen te wijzigen die zijn gemaakt door AUTO CDC ... INTO instructies.

Omdat de pijplijn een opgegeven SEQUENCE BY kolom gebruikt en de juiste sequentiërende waarden doorgeeft aan de __START_AT en __END_AT kolommen van de doeltabel (voor SCD-type 2), moet u ervoor zorgen dat DML-instructies geldige waarden voor deze kolommen gebruiken om de juiste volgorde van records te behouden. Zie Hoe AUTO CDC werkt.

Zie Gegevens toevoegen, wijzigen of verwijderen in een streamingtabel voor meer informatie over het gebruik van DML-instructies met streamingtabellen.

In het volgende voorbeeld wordt een actieve record ingevoegd met een beginvolgorde van 5:

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;

Een wijzigingsgegevensstroom lezen uit een AUTO CDC-doeltabel

In Databricks Runtime 15.2 en hoger kunt u een gegevensfeed voor wijzigingen lezen uit een streamingtabel die het doel is van AUTO CDC of AUTO CDC FROM SNAPSHOT query's op dezelfde manier als u een gegevensfeed van andere Delta-tabellen leest. Het volgende is vereist om de wijzigingsgegevensfeed te kunnen lezen uit een streamingdoeltabel.

• De doelstreamingtabel moet worden gepubliceerd naar Unity Catalog. Zie Unity Catalog gebruiken met pijplijnen.
• Als u de wijzigingengegevensfeed wilt lezen uit de doelstreamingtabel, moet u Databricks Runtime 15.2 of hoger gebruiken. Als u de wijzigingengegevensfeed in een andere pijplijn wilt lezen, moet de pijplijn zijn geconfigureerd voor het gebruik van Databricks Runtime 15.2 of hoger.

U leest de wijzigingengegevensfeed uit een doelstreamingtabel die in een Lakeflow-pijplijn is gemaakt, op dezelfde manier als het lezen van een wijzigingengegevensfeed uit andere Delta-tabellen. Zie Change data feed gebruiken in Azure Databricks voor meer informatie over het gebruik van de functionaliteit voor Delta change data feed, inclusief voorbeelden in Python en SQL.

Lees een wijzigingsdatafeed vanuit een gematerialiseerde weergave

Je kunt een wijzigingsdatafeed lezen vanuit een gematerialiseerde weergave die is gemaakt in een Lakeflow-pijplijn of in Databricks SQL. Gebruik dit om gematerialiseerde weergavewijzigingen te repliceren naar bestemmingen buiten Azure Databricks, of om een geschiedenis bij te houden van gematerialiseerde weergavewijzigingen voor audit en rapportage.

Gematerialiseerde weergaven gebruiken een automatische wijzigingsdatafeed, dus je zet de change data-feed zelf niet in. In plaats daarvan schakel je de wijzigingsdatafeed in op elke gematerialiseerde weergave waar je het nodig hebt door aan de volgende eisen te voldoen. Zie Automatische wijzigingsgegevensfeed.

• Om de feed met wijzigingsgegevens te lezen, moet je Databricks Runtime 18 LTS of hoger gebruiken op classic compute, serverless compute of Databricks SQL.

• De gematerialiseerde weergave, de pijplijn die het aanmaakt, of de pijplijn die het leest, moeten het PREVIEW kanaal gebruiken.

• Voor de gematerialiseerde weergave moet rijtracering zijn ingeschakeld. Gematerialiseerde weergaven op serverloze rekenkracht hebben tracering van rijen standaard ingeschakeld. Zie Het bijhouden van rijen in Azure Databricks. Om te controleren of rijtracking is ingeschakeld in een gematerialiseerde weergave, voer je uit:

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Om de feed met wijzigingsgegevens uit een gematerialiseerde weergave te lezen, schakelt u de markering voor externe metagegevens in voor de pipeline of de gematerialiseerde weergave. Voor instructies, zie Hoe je toegang voor een dataset inschakelt.

Je leest de wijzigingsdatafeed uit een gematerialiseerde weergave op dezelfde manier als uit andere Delta-tabellen, met behulp van de table_changes() functie, een streaming read of de readChangeFeed optie. Voor syntaxis en voorbeelden in SQL en Python, zie Use change data feed op Azure Databricks.

Je kunt een feed voor wijzigingsgegevens van een gematerialiseerde weergave lezen in een gematerialiseerde weergave of streamingtabel van Databricks SQL:

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

Limitations

Naast de beperkingen van automatische wijzigingsgegevensfeeds gelden de volgende wanneer u een wijzigingsdatafeed leest vanuit een gematerialiseerde weergave:

• De wijzigingsdatafeed bevat ongewijzigde rijen wanneer de gematerialiseerde weergave volledig is herschreven, en het consolideert niet meerdere updates van dezelfde rij in één gebeurtenis. Om deze eruit te filteren, aggregeer je de wijzigingsdatafeed door ze op alle kolommen te groeperen om inserts en deletes te vinden die dezelfde rijwaarden delen.
• Alleen Azure Databricks kan de change data-feed opvragen voor een gematerialiseerde weergave. Externe klanten van Delta Lake en Iceberg kunnen dat niet.
• In Lakeflow-pijplijnen kun je de change data feed van een gematerialiseerde weergave alleen lezen vanuit een andere pijplijn, en die pijplijn moet het kanaal PREVIEW gebruiken. Het lezen van de feed met wijzigingsgegevens van een gematerialiseerde weergave in dezelfde pipeline die deze maakt, wordt niet ondersteund.
• Je kunt geen vectorzoekindex maken vanuit een gematerialiseerde weergave.

Verkrijg gegevens over records die door een CDC-query in pijplijnen worden verwerkt

De volgende metrische gegevens worden vastgelegd door AUTO CDC query's:

• num_upserted_rows: Het aantal uitvoerrijen dat tijdens een update in de gegevensset is geüpsert.
• num_deleted_rows: Het aantal bestaande uitvoerrijen dat tijdens een update uit de gegevensset is verwijderd.

De num_output_rows metrieke, output voor niet-CDC-flows, wordt niet vastgelegd voor AUTO CDC queries.

Gedeeltelijke updates toepassen

Wanneer een bron alleen de gewijzigde kolommen verzendt, AUTO CDC moet u onderscheid maken tussen een kolom die ontbreekt in een wijzigingsrecord, waardoor de doelwaarde ongewijzigd moet blijven en een kolom die expliciet is ingesteld nullop , waarmee de doelwaarde moet worden overschreven.null Standaard behandelt IGNORE NULL UPDATES elke null als een markering voor 'niet bijwerken', waardoor het geen expliciete null kan toepassen. Als u deze dubbelzinnigheid wilt oplossen, kiest u een van de volgende drie methoden:

COLUMNS TO UPDATE kan niet worden gecombineerd met IGNORE NULL UPDATES, en wordt niet ondersteund voor bitemporale tabellen.

Kies bij wijze van vuistregel voor COLUMNS TO UPDATE wanneer de producent weet welke kolommen in elk record zijn gewijzigd en die informatie kan opnemen in een bronkolom, bijvoorbeeld wanneer meerdere producenten naar dezelfde gegevensbron schrijven of de set bijwerkbare kolommen in de loop van de tijd groter wordt. Kies IGNORE NULL UPDATES ON wanneer de eigenaar van de pijplijn de vaste set van vooraf in te stellen kolommen kent en deze liever beheert in pijplijncode.

In het volgende voorbeeld wordt een bronkolom met de naam gebruikt columnsToUpdate om te bepalen welke kolommen elke recordupdate wijzigen, inclusief kolommen die expliciet nullzijn ingesteld op:

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;

Zie AUTO CDC INTO (pijplijnen) en create_auto_cdc_flow voor de volledige parameterreferentie.

Bitemporele AUTO CDC

SCD Type 1 en Type 2 zijn unitemporal: ze volgen wijzigingen in één dimensie. Bitemporal breidt de SCD Type 2-geschiedenis uit om wijzigingen in twee tijdsdimensies bij te houden en onderscheid te maken tussen twee perspectieven:

• Zakelijke tijd: wanneer de gebeurtenis daadwerkelijk is gebeurd.
• Systeemtijd: wanneer het systeem de gebeurtenis heeft geregistreerd of opgenomen.

Net als SCD Type 2 behoudt bitemporal een volledige geschiedenis van records. Er wordt een tweede tijdlijn toegevoegd, zodat u zowel kunt reconstrueren wat de gegevens hebben laten zien als wat het systeem in het verleden geloofde.

Een hedgefonds neemt bijvoorbeeld voorraadgegevens op uit een bronsysteem. De aandelenkoers van Acme Corp verandert op 1 januari, maar het fonds verwerkt die update pas op 5 januari. Bitemporal AUTO CDC laat het fonds twee verschillende vragen beantwoorden: wat de werkelijke aandelenkoers van Acme Corp was op 1 januari (bedrijfstijd) en welke prijs het systeem geloofde toen het fonds handelsbeslissingen maakte op 3 januari (systeemtijd). De mogelijkheid om onderscheid te maken tussen deze tijdlijnen is handig voor controle, regelgevingsrapportage en financiële besluitvorming.

Als u bitemporale verwerking wilt inschakelen, stelt u STORED AS BITEMPORAL (SQL) of stored_as_scd_type="bitemporal" (Python) in, gebruikt u SEQUENCE BY voor de bedrijfstijdkolom en gebruikt u SYSTEM SEQUENCE BY voor de systeemtijdkolom. De doeltabel voegt de kolommen __SYSTEM_START_AT en __SYSTEM_END_AT toe, naast de SCD Type 2-kolommen __START_AT en __END_AT. Zie AUTO CDC INTO (pijplijnen) of create_auto_cdc_flow voor syntaxisdetails.

Voorbeelden van bitemporal AUTO CDC

In het volgende voorbeeld wordt een bitemporale doeltabel gemaakt op basis van een kleine set synthetische CDC-gebeurtenissen. De bt kolom heeft de bedrijfstijd en de st kolom heeft de systeemtijd.

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;

De volgende reeks wijzigingen laat zien hoe een bitemporale tabel een invoegbewerking, een update, een out-of-order-update en een verwijdering voor één bedrijf registreert. De kolom voor sequentiëren genereert de __START_AT kolommen en __END_AT kolommen (bedrijfstijd) en de kolom voor systeemvolgorde genereert de __SYSTEM_START_AT kolommen en __SYSTEM_END_AT (systeemtijd):

Het systeem verwerkt gebeurtenissen die in elke volgorde in beide tijdlijnen binnenkomen. Wanneer een gebeurtenis binnenkomt met een eerdere bedrijfstijd of systeemtijd dan gebeurtenissen die al zijn verwerkt, corrigeert het systeem de betreffende geschiedenis in plaats van alleen aan het einde toe te voegen.

Wijziging 1: Invoegen

Bedrijf A wordt toegevoegd op 18-7-2025 10:01:00 (kantoortijd) maar wordt niet opgenomen tot 10:05:00 (systeemtijd).

Invoer:

Uitvoer:

XFv1 is geldig vanaf 10:01:00 zonder bekend einde. Het systeem heeft dit feit geleerd op systeemtijd 10:05:00, zonder bekend einde.

Wijziging 2: Bijwerken

Bedrijf A wordt bijgewerkt om 18-07-2025 12:15:43 (bedrijfstijd) en het systeem verwerkt de gebeurtenis om 12:20:00 (systeemtijd). Het systeem behoudt zowel wat het geloofde voordat de update bekend was en de gecorrigeerde bedrijfsgeschiedenis nadat de update is opgenomen.

Invoer:

Uitvoer:

XFv1 werd verondersteld geldig te zijn van 10:01:00 zonder bekend einde, en het systeem hield dat geloof van 10:05:00 tot 12:20:00 uur. XFv1 is nu alleen geldig tot 12:15:43, een gecorrigeerde geschiedenis vanaf systeemtijd 12:20:00 zonder bekend einde. XFv2 is geldig vanaf 12:15:43 zonder bekend einde en is geleerd op systeemtijd 12:20:00.

Wijziging 3: Update buiten volgorde

Er wordt een update buiten bestelling ontvangen die aangeeft dat bedrijf A daadwerkelijk is bijgewerkt op 18-7-2025 12:05:00 (kantoortijd), maar het wordt pas opgenomen tot 12:25:00 (systeemtijd). Wanneer een update later binnenkomt in de systeemtijd, maar met een voorgaande bedrijfstijd, corrigeert het systeem de historische bedrijfstijd en behoudt het zowel wat het geloofde vóór de out-of-order-update als de gecorrigeerde geschiedenis.

Invoer:

Uitvoer:

XFv1 werd verondersteld geldig te zijn van 10:01:00 tot 12:15:43, en dat geloof is nu geldig in systeemtijd tot 12:25:00. De nieuwe update corrigeert de zakelijke geldigheid van XFv1 zodat deze eindigt om 12:05:00, met een gecorrigeerde geschiedenis die van kracht is vanaf systeemtijd 12:25:00. XFv3 is nu bekend om geldig te zijn van 12:05:00 tot 12:15:43, een overtuiging geldig in systeemtijd van 12:25:00 zonder bekend einde.

Wijziging 4: Verwijderen

Bedrijf A is verwijderd op 18-07-2025 om 12:30:00, en het systeem verwerkt de gebeurtenis om 12:30:00. Omdat een verwijderbewerking het einde van het bedrijfsleven van de entiteit aangeeft, maakt het systeem geen vervangingsrij. XFv2 verschijnt in twee rijen, waarbij een volledige audittrail behouden blijft van zowel het moment waarop het bedrijf ophield te bestaan als het moment waarop het systeem kennisnam van de verwijdering.

Invoer:

Uitvoer:

XFv2 was geldig vanaf 12:15:43 zonder bekend eindtijdstip, en het systeem ging daarvan uit van 12:20:00 tot 12:30:00. Nadat de verwijdering is verwerkt, is van XFv2 bekend dat deze alleen geldig is tot 12:30:00, een gecorrigeerde geschiedenis die ingaat vanaf systeemtijd 12:30:00.

Welke gegevensobjecten worden gebruikt voor CDC-verwerking in een pijplijn?

Wanneer u de doeltabel in de Hive-metastore opgeeft, worden er twee gegevensstructuren aangemaakt.

• Een weergave met de naam die is toegewezen aan de doeltabel.
• Een interne back-uptabel die door de pijplijn wordt gebruikt voor het beheren van CDC-verwerking. De naam van deze tabel wordt gevormd door __apply_changes_storage_ voor de naam van de doeltabel te plaatsen.

Als u bijvoorbeeld een doeltabel met de naam dp_cdc_targetdeclareert, ziet u een weergave met de naam dp_cdc_target en een tabel met de naam __apply_changes_storage_dp_cdc_target in de metastore. Query's uitvoeren op de weergave om toegang te krijgen tot de verwerkte gegevens. Wijzig de backingtabel niet rechtstreeks.