De AUTO CDC-API's: het vastleggen van wijzigingsgegevens vereenvoudigen met pijplijnen

Lakeflow-pijplijnen vereenvoudigen het vastleggen van wijzigingsgegevens (CDC) met de AUTO CDC en AUTO CDC FROM SNAPSHOT API's. Deze API's automatiseren de complexiteit van het berekenen van langzaam veranderende dimensies (SCD) Type 1 en Type 2 vanuit een CDC-feed of databasemomentopnamen. De AUTO CDC API biedt ook ondersteuning voor bitemporale tracering, waarmee wijzigingen worden vastgelegd in twee tijddimensies (bèta). Raadpleeg Wijzigingsgegevens vastleggen en momentopnamen voor meer informatie over SCD Type 1 en Type 2. Zie Bitemporal AUTO CDC voor meer informatie over bitemporale tracering.

Opmerking

De AUTO CDC API's vervangen de APPLY CHANGES API's en hebben dezelfde syntaxis. De APPLY CHANGES API's zijn nog steeds beschikbaar, maar Databricks raadt het gebruik van de AUTO CDC API's aan.

De API die u gebruikt, is afhankelijk van de bron van uw wijzigingsgegevens:

  • AUTO CDC: Gebruik deze optie wanneer voor de brondatabase een CDC-feed is ingeschakeld. AUTO CDC verwerkt wijzigingen uit een gegevenswijzigingenfeed (CDF). Het wordt ondersteund in zowel de SQL-pijplijn- als Python-interfaces.
  • AUTO CDC FROM SNAPSHOT: Gebruik deze optie wanneer CDC niet is ingeschakeld voor de brondatabase en alleen momentopnamen beschikbaar zijn. Met deze API worden momentopnamen vergeleken om wijzigingen te bepalen en vervolgens worden ze verwerkt. Het wordt ondersteund in zowel de SQL-pijplijn- als Python-interfaces.

Voor SCD Type 1-doelen kun je ook een AUTO CDC FROM SNAPSHOT flow combineren met één of meer AUTO CDC flows. Gebruik dit patroon wanneer een bron periodieke gezaghebbende snapshots en een CDC-feed met lagere latentie tussen snapshots levert.

Beide API's ondersteunen het bijwerken van tabellen met SCD-type 1 en Type 2:

  • Gebruik SCD Type 1 om records rechtstreeks bij te werken. Geschiedenis wordt niet bewaard voor bijgewerkte gegevens.
  • Gebruik SCD Type 2 om een geschiedenis van records te bewaren, hetzij op alle soorten updates of op updates voor een opgegeven set kolommen.

Voor AUTO CDC alleen kunt u ook bitemporale opslag gebruiken, waarmee de SCD Type 2-geschiedenis wordt uitgebreid om wijzigingen in twee dimensies bij te houden: bedrijfstijd en systeemtijd. Bitemporal zit in Beta. Zie Bitemporal AUTO CDC.

AUTO CDC ondersteunt ook gedeeltelijke updates, waarbij een wijzigingsrecord slechts een subset kolommen bijwerken. Zie Gedeeltelijke updates toepassen.

De AUTO CDC API's worden niet ondersteund door declaratieve Apache Spark-pijplijnen.

Zie AUTO CDC INTO (pijplijnen),create_auto_cdc_flow en create_auto_cdc_from_snapshot_flow voor syntaxis en andere verwijzingen.

Opmerking

Gebruik deze API's om tabellen in je pijplijnen bij te werken op basis van wijzigingen in brondata. Zie Feed voor wijzigingsgegevens gebruiken in Azure Databricks voor meer informatie over het vastleggen en opvragen van wijzigingsgegevens op rijniveau voor Delta-tabellen.

Requirements

Als u de CDC-API's wilt gebruiken, moet uw pijplijn geconfigureerd zijn om serverloze Lakeflow-pijplijnen of de Lakeflow-pijplijnen Pro of Advancededities te gebruiken.

Hoe AUTO CDC werkt

Als u CDC-verwerking wilt uitvoeren, maakt u een streamingtabel met AUTO CDC en gebruikt u vervolgens de AUTO CDC ... INTO-instructie in SQL of de create_auto_cdc_flow()-functie in Python om de bron, sleutels en volgordebepaling voor de wijzigingenfeed op te geven. Zie Gegevens vastleggen en momentopnamen wijzigen voor een uitleg van de werking van sequentiëren en SCD-logica. Zie de AUTO CDC-voorbeelden.

Voor initiële hydratatie vanuit een bron met een change feed, maakt u gebruik van AUTO CDC met een once-stroom en vervolgt u het verwerken van de change feed. Zie Een externe RDBMS-tabel repliceren met behulp van AUTO CDC.

Zie AUTO CDC INTO (pijplijnen) of create_auto_cdc_flow voor syntaxisdetails.

Hoe AUTO CDC FROM SNAPSHOT werkt

AUTO CDC FROM SNAPSHOT bepaalt wijzigingen in brongegevens door momentopnamen in volgorde te vergelijken. U kunt momentopnamen uit een Delta-tabel, cloudopslagbestanden of JDBC rechtstreeks lezen. Het wordt ondersteund in zowel de SQL-pijplijn- als Python-interfaces.

Om CDC-verwerking uit te voeren met AUTO CDC FROM SNAPSHOT, maak een streamingtabel aan en definieer vervolgens de flow:

  • Gebruik in SQL de AUTO CDC ... FROM SNAPSHOT vorm van de CREATE FLOW statement of de ingebedde FLOW AUTO CDC clausule van CREATE STREAMING TABLE. De bron is een vereiste FROM SNAPSHOT (snapshot_query) clausule en een optionele WITH VERSION (version_query) clausule die de volgende snapshotversie selecteert om te verwerken. Zie CREATE FLOW (pipelines).
  • In Python gebruik je de create_auto_cdc_from_snapshot_flow() functie om de snapshot, sleutels en andere argumenten te specificeren. Zie create_auto_cdc_from_snapshot_flow.

Zie Momentopnameverwerkingspatronen voor meer informatie over de twee innamepatronen en wanneer u elk moet gebruiken. Zie de voorbeelden van AUTO CDC FROM SNAPSHOT.

Combineer AUTO CDC FROM SNAPSHOT en AUTO CDC

De snapshotversie en de CDC-sequencingkolom vormen één ordeningsdomein. Een CDC-gebeurtenis die nieuwer is dan de laatste snapshot heeft voorrang boven de snapshot. Een nieuwere snapshot gaat voor oudere CDC-gebeurtenissen en behandelt sleutels die ontbreken in de snapshot als verwijderd bij de snapshotversie. Nieuwere CDC-evenementen kunnen die sleutels behouden of herstellen.

Een geünificeerd doel moet aan de volgende eisen voldoen:

  • Het doel gebruikt SCD Type 1.
  • Het target heeft precies één AUTO CDC FROM SNAPSHOT stroom en één of meer AUTO CDC stromen.
  • Elke AUTO CDC flow heeft een unieke naam.
  • Alle flows gebruiken hetzelfde aantal keys in dezelfde volgorde. Snapshot-flow sleutelnamen worden zonder onderscheid tussen hoofdletters en kleine letters vergeleken met AUTO CDC sleutelnamen. Meerdere stromen AUTO CDC moeten identieke sleutelnamen en hoofdlettergebruik gebruiken.
  • De snapshotversie en elke CDC-sequencingkolom hebben exact hetzelfde datatype.
  • De AUTO CDC FROM SNAPSHOT flow definieert geen verwachtingen.
  • De AUTO CDC flows gebruiken IGNORE NULL UPDATES of de kolomlijstvarianten ervan niet.
  • De pijplijn gebruikt de getriggerde modus.
  • Als je een AUTO CDC flow toevoegt aan een bestaand AUTO CDC FROM SNAPSHOT doel, mag het doel geen gebruikerskolommen bevatten waarvan de namen conflicteren met gereserveerde AUTO CDC systeemkolommen.

Zowel AUTO CDC als AUTO CDC FROM SNAPSHOT flows kunnen de Python- of SQL-pijplijninterface gebruiken. Je kunt SQL- en Python-flows combineren in dezelfde target.

Gebruik once=True in Python of ONCE in SQL op de AUTO CDC FROM SNAPSHOT flow om een snapshot één keer te backfillen terwijl de AUTO CDC flows nieuwe gebeurtenissen bij volgende updates blijven verwerken. Een volledige verversing van de target voert de eenmalige snapshotflow opnieuw uit. Voor een volledig voorbeeld, zie Voeg een backfill toe aan een AUTO CDC SCD Type 1 tabel.

Dit patroon ondersteunt geen SCD Type 2- of bitemporele doeltabellen.

Meerdere kolommen gebruiken voor sequentiëren

Als u wilt sorteren op meerdere kolommen (bijvoorbeeld een tijdstempel en een ID om de volgorde te bepalen), gebruikt u een STRUCT om deze te combineren. De API sorteert eerst op het eerste veld, en in het geval van een gelijke waarde, op het tweede veld, enzovoort.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

sequence_by = struct("timestamp_col", "id_col")

Auto CDC-voorbeelden

In de volgende voorbeelden ziet u de verwerking van SCD Type 1 en Type 2 met behulp van een gegevensfeedbron voor wijzigingen. Met de voorbeeldgegevens worden nieuwe gebruikersrecords gemaakt, een gebruikersrecord verwijderd en worden gebruikersrecords bijgewerkt. In het SCD-type 1-voorbeeld komen de laatste UPDATE bewerkingen te laat en worden ze verwijderd uit de doeltabel, waarmee de verwerking van gebeurtenissen buiten de order wordt gedemonstreerd.

Hier volgen de invoerrecords die in deze voorbeelden worden gebruikt. Deze gegevens worden gemaakt door de query uit te voeren in de sectie Voorbeeldgegevens maken .

userId naam city operatie volgordenummer
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lelie Cancun INSERT 2
123 Nul Nul Verwijderen 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Als u de opmerkingen bij de laatste rij in de query voor het genereren van voorbeeldgegevens ongedaan maakt, wordt de volgende record ingevoegd die aangeeft dat de tabel moet worden afgekapt (wis de tabel) op sequenceNum=3:

userId naam city operatie volgordenummer
Nul Nul Nul AFKNOTTEN 3

Opmerking

Alle volgende voorbeelden bevatten opties om zowel als DELETETRUNCATE bewerkingen op te geven, maar elk is optioneel.

Voorbeeldgegevens maken

Voer de volgende instructies uit om een voorbeeldgegevensset te maken. Deze code is niet bedoeld om te worden uitgevoerd als onderdeel van een pijplijndefinitie. Voer deze uit vanuit de verkenningsmap van uw pijplijn in plaats van de map transformaties.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

SCD Type 1-updates verwerken

SCD Type 1 bewaart alleen de nieuwste versie van elke record. In het volgende voorbeeld wordt de bovenstaande wijzigingengegevensstroom gelezen en worden wijzigingen toegepast op een streamingtabeldoel. Je hebt een pipeline nodig om deze code uit te voeren. Zie ETL-pijplijnen ontwikkelen en fouten opsporen met de Lakeflow Pipelines Editor.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Nadat het SCD-type 1-voorbeeld is uitgevoerd, bevat de doeltabel de volgende records:

userId naam city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lelie Cancun

Gebruiker 123 (Isabel) is verwijderd en wordt niet weergegeven. Gebruiker 125 (Mercedes) toont alleen de meest recente stad (Guadalajara) omdat SCD Type 1 eerdere waarden overschrijft. De eerdere UPDATE bij sequenceNum=5 is verwijderd omdat er later een update bij sequenceNum=6 arriveerde.

Nadat u het voorbeeld hebt uitgevoerd met de TRUNCATE record niet van commentaar voorzien, wordt bij sequenceNum=3 de tabel gewist. Dit betekent dat records 124 en 126 zich niet in de tabel bevinden en dat de uiteindelijke doeltabel alleen de volgende record bevat:

userId naam city
125 Mercedes Guadalajara

SCD Type 2-updates verwerken

SCD Type 2 behoudt een volledige geschiedenis van wijzigingen door nieuwe rijen te maken voor elke versie van een record, met __START_AT en __END_AT kolommen die aangeven wanneer elke versie actief was.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Nadat het SCD-type 2-voorbeeld is uitgevoerd, bevat de doeltabel de volgende records:

userId naam city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 Nul
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 Nul
126 Lelie Cancun 2 Nul

De tabel behoudt de volledige geschiedenis. Gebruiker 123 heeft twee versies (beëindigd op reeks 6 wanneer deze is verwijderd). Gebruiker 125 heeft drie versies met stadswijzigingen. Records met __END_AT = null zijn momenteel actief.

Een kolomsubset bijhouden met SCD Type 2

SCD Type 2 maakt standaard een nieuwe versie wanneer een kolomwaarde wordt gewijzigd. U kunt een subset van kolommen opgeven die u wilt bijhouden, zodat wijzigingen in andere kolommen de huidige versie bijwerken in plaats van een nieuwe geschiedenisrecord te genereren.

In het volgende voorbeeld wordt de city kolom uitgesloten van het bijhouden van geschiedenis:

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Omdat city wijzigingen niet worden bijgehouden, overschrijven stadsupdates de huidige rij in plaats van een nieuwe versie te maken. De doeltabel bevat de volgende records:

userId naam city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 Nul
125 Mercedes Guadalajara 2 Nul
126 Lelie Cancun 2 Nul

VOORBEELDEN VAN AUTO CDC FROM SNAPSHOT

De volgende secties bieden voorbeelden van het gebruik van AUTO CDC FROM SNAPSHOT om momentopnamen te verwerken tot SCD Type 1- of Type 2-doeltabellen. Zie Gegevens vastleggen en momentopnamen wijzigen voor achtergrondinformatie over het gebruik van deze API.

De voorbeelden in deze sectie bevatten Python- en SQL-implementaties. Voor volledige SQL-syntaxis, zie CREATE FLOW (pipelines).

Voorbeeld: Periodieke snapshots verwerken

Gebruik deze aanpak wanneer snapshots regelmatig en in volgorde binnenkomen. De Python-interface gebruikt de pipeline-updatevolgorde voor versiebeheer. Voor SQL gebruikt het voorbeeld een aparte versietabel om te identificeren wanneer er een nieuwe snapshot beschikbaar is.

U kunt momentopnamen van meerdere brontypen lezen, waaronder Delta-tabellen, cloudopslagbestanden en JDBC-verbindingen.

Stap 1: Voorbeeldgegevens maken

Maak een tabel met momentopnamegegevens. Voer de volgende code uit vanuit een notebook of Databricks SQL in de explorations map van uw pijplijn:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Stap 2: AUTO CDC UITVOEREN VANAF MOMENTOPNAME

Je hebt een pipeline nodig om de code in deze stap uit te voeren. Zie ETL-pijplijnen ontwikkelen en fouten opsporen met de Lakeflow Pipelines Editor.

Voor Python kies je een brontype voor de snapshotweergave (de voorbeeldcreatiecode genereert een Delta-tabel):

Optie A: Lezen uit een Delta-tabel 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Optie B: Lezen uit cloudopslag 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Optie C: Lezen uit JDBC (alleen klassiek berekenen) 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Voeg dan de doeltabel en de flow toe. De SQL-implementatie gebruikt de snapshot_version tabel om te identificeren wanneer er een nieuwe snapshot beschikbaar is:

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Na de eerste pijplijnuitvoering worden alle records als actieve rijen ingevoegd.

userId city __START_AT __END_AT
1 Oaxaca 0 Nul
2 Monterrey 0 Nul
3 Tijuana 0 Nul

Opmerking

Om in plaats daarvan SCD Type 1 te gebruiken en alleen de huidige status te behouden, stel stored_as_scd_type=1 in Python in of STORED AS SCD TYPE 1 in SQL. In dit geval bevat de doeltabel de kolommen __START_AT en __END_AT niet.

Stap 3: Een nieuwe momentopname simuleren en opnieuw uitvoeren

Werk de brontabel bij om een nieuwe snapshot te simuleren (voer deze code uit vanuit een notebook of SQL-bestand in de explorations map van je pipeline):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Voer de pijplijn opnieuw uit. AUTO CDC FROM SNAPSHOT vergelijkt de nieuwe momentopname met de vorige en detecteert dat gebruiker 1 is verwijderd, gebruikers 2 en 3 zijn bijgewerkt en gebruikers 4 en 6 zijn ingevoegd. Hiermee wordt een wijzigingenfeed gegenereerd en gebruikt AUTO CDC om de uitvoertabel te maken.

Na de tweede iteratie met SCD Type 2 bevat de doeltabel de volgende gegevens:

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 Nul
3 Tijuana 0 1
3 Los Angeles 1 Nul
4 Death Valley 1 Nul
6 Kings Canyon 1 Nul

Gebruiker 1 is beëindigd (verwijderd). Gebruikers 2 en 3 hebben elk twee versies met hun stadswijzigingen. Gebruikers 4 en 6 zijn nieuw ingevoegd.

Na de tweede uitvoering met SCD Type 1 geeft de doeltabel alleen de huidige status weer:

userId city
2 Carmel
3 Los Angeles
4 Death Valley
6 Kings Canyon

Voorbeeld: Momentopnamen verwerken met behulp van versiefuncties

Gebruik deze methode wanneer u expliciet controle nodig hebt over het ordenen van momentopnamen. Gebruik deze benadering bijvoorbeeld wanneer meerdere momentopnamen tegelijkertijd aankomen of momentopnamen niet in de volgorde komen. Je bepaalt hoe je de volgende snapshot en het bijbehorende versienummer selecteert. De API verwerkt momentopnamen in oplopende versievolgorde:

  • Als meerdere snapshots in opslag zijn, verwerkt de voorbeeldselector beschikbare snapshots in versievolgorde.
  • Als een snapshot uit de volgorde aankomt (bijvoorbeeld snapshot_3 na snapshot_4), sluit de voorbeeldselector deze uit omdat de versie niet groter is dan de laatst gecommitteerde versie.
  • Als er geen nieuwe snapshots zijn, geeft de versiefunctie of query geen resultaat en vindt er geen verwerking plaats.

Stap 1: Momentopnamebestanden voorbereiden

Maak CSV-bestanden met momentopnamegegevens en voeg ze toe aan een volume- of cloudopslaglocatie. Geef de bestanden chronologisch een naam (bijvoorbeeld snapshot_1.csv, snapshot_2.csv).

Elk bestand moet kolommen bevatten voor userId en city. Voorbeeld:

snapshot_1.csv:

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId city
2 Carmel
3 Los Angeles
4 Death Valley

Stap 2: AUTO CDC FROM SNAPSHOT uitvoeren met een versiefunctie

Maak in een pijplijn een nieuw Python- of SQL-bestand aan in de transformations map en voeg de volgende code toe. Voer vervolgens de pijplijn uit. Zie ETL-pijplijnen ontwikkelen en fouten opsporen met de Lakeflow Pipelines Editor.

Python

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Opmerking

Om in plaats daarvan SCD Type 1 te gebruiken, stel stored_as_scd_type=1 in Python in of STORED AS SCD TYPE 1 in SQL.

Na de verwerking snapshot_1.csv bevat de doeltabel de volgende records:

userId city __START_AT __END_AT
1 Oaxaca 1 Nul
2 Monterrey 1 Nul
3 Tijuana 1 Nul

Na de verwerking snapshot_2.csv bevat de doeltabel de volgende records:

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 Nul
3 Tijuana 1 2
3 Los Angeles 2 Nul
4 Death Valley 2 Nul

Opmerking

Houd er rekening mee dat voor SCD Type 1 de tabel er precies zo uitziet als de meest recente momentopname. Het verschil is dat downstreamquery's de wijzigingenfeed kunnen gebruiken om alleen gewijzigde records te verwerken.

Stap 3: Nieuwe momentopnamen toevoegen

Voeg een nieuw CSV-bestand toe aan de opslaglocatie met gewijzigde gegevens (bijvoorbeeld gewijzigde plaatswaarden, nieuwe rijen of verwijderde rijen). Voer vervolgens de pijplijn opnieuw uit om de nieuwe momentopname te verwerken.

Beperkingen

  • De kolom voor sequentiëren moet een sorteerbaar gegevenstype zijn. NULL Sequentiërende waarden worden niet ondersteund.
  • Als u gegevens wilt streamen van het doel van een AUTO CDC-proces, leest u deze uit de wijzigingenfeed. Voor meer informatie, zie Een wijzigingsgegevensfeed lezen uit een AUTO CDC-doeltabel.

Aanvullende bronnen

  • Laatst bijgewerkt om