create_auto_cdc_from_snapshot_flow

Met create_auto_cdc_from_snapshot_flow de functie maakt u een stroom die gebruikmaakt van de functie Lakeflow-pijplijnen voor het wijzigen van gegevensopnamen (CDC) om brongegevens van databasemomentopnamen te verwerken. Zie hoe AUTO CDC FROM SNAPSHOT werkt.

Opmerking

Deze functie vervangt de vorige functie apply_changes_from_snapshot(). De twee functies hebben dezelfde handtekening. Databricks raadt het bijwerken aan om de nieuwe naam te gebruiken.

Belangrijk

Voor deze bewerking moet u een gerichte streamingstabel hebben. Als u de vereiste doeltabel wilt maken, kunt u de functie create_streaming_table() gebruiken.

Syntaxis

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Opmerking

Voor AUTO CDC FROM SNAPSHOT verwerking is het standaardgedrag het invoegen van een nieuwe rij wanneer een overeenkomende record met dezelfde sleutel(en) niet in het doel bestaat. Als er wel een overeenkomende record bestaat, wordt deze alleen bijgewerkt als een van de waarden in de rij is gewijzigd. Rijen met sleutels die aanwezig zijn in de doeldatabase, maar die niet meer aanwezig zijn in de brondatabase, worden verwijderd.

Zie de AUTO CDC-API's voor meer informatie over CDC-verwerking met momentopnamen: Het vastleggen van wijzigingsgegevens vereenvoudigen met pijplijnen. Voor voorbeelden van het gebruik van de create_auto_cdc_from_snapshot_flow() functie, zie de periodieke snapshot-opname, historische snapshot-opname en SCD Type 1 backfill-voorbeelden .

Parameterwaarden

Kenmerk	Typologie	Description
`target`	`str`	Verplicht. De naam van de tabel die moet worden bijgewerkt. U kunt de functie create_streaming_table() gebruiken om de doeltabel te maken voordat u de `create_auto_cdc_from_snapshot_flow()` functie uitvoert.
`source`	`str` of `lambda function`	Verplicht. Ofwel de naam van een tabel of weergave om periodiek een momentopname te maken of een Python-lambda-functie die de dataframe voor momentopnamen retourneert die moet worden verwerkt en de versie van de momentopname. Zie Het argument implementeren`source`.
`keys`	`list`	Verplicht. De kolom of combinatie van kolommen waarmee een rij in de brongegevens uniek wordt geïdentificeerd. Dit wordt gebruikt om te bepalen welke CDC-gebeurtenissen van toepassing zijn op specifieke records in de doeltabel. U kunt een van de volgende opties opgeven: Een lijst met tekenreeksen: `["userId", "orderId"]` Een lijst met Spark SQL-functies `col()` : `[col("userId"), col("orderId"]`. Argumenten voor `col()` functies kunnen geen kwalificaties bevatten. U kunt bijvoorbeeld gebruiken `col(userId)`, maar u kunt het niet gebruiken `col(source.userId)`.
`stored_as_scd_type`	`str` of `int`	Om te bepalen of records moeten worden opgeslagen als SCD-type 1 of SCD-type 2. Ingesteld op `1` voor SCD-type 1 of `2` voor SCD-type 2. De standaardwaarde is SCD type 1.
`track_history_column_list` of `track_history_except_column_list`	`list`	Een subset van uitvoerkolommen om bij te houden voor historische doeleinden in de doeltabel. Hiermee `track_history_column_list` geeft u de volledige lijst met kolommen op die moeten worden bijgehouden. Hiermee `track_history_except_column_list` geeft u de kolommen op die moeten worden uitgesloten van het bijhouden. U kunt een waarde declareren als een lijst met tekenreeksen of als Spark SQL-functies `col()` : `track_history_column_list = ["userId", "name", "city"]` `track_history_column_list = [col("userId"), col("name"), col("city")]` `track_history_except_column_list = ["operation", "sequenceNum"]` `track_history_except_column_list = [col("operation"), col("sequenceNum")` Argumenten voor `col()` functies kunnen geen kwalificaties bevatten. U kunt bijvoorbeeld gebruiken `col(userId)`, maar u kunt het niet gebruiken `col(source.userId)`. De standaardinstelling is om alle kolommen in de doeltabel op te nemen wanneer er geen `track_history_column_list` of `track_history_except_column_list` argument wordt doorgegeven aan de functie.
`once`	`bool`	Of je de snapshotflow slechts één keer moet uitvoeren. Wanneer `True`, wordt de flow overgeslagen tijdens opeenvolgende incrementele updates nadat deze succesvol is gecomm. Een volledige verversing van het doel laat de flow opnieuw lopen. Gebruik deze optie voor een eenmalige snapshot-load of backfill. De standaardwaarde is `False`.

Notes

Voor SCD Type 1-doelen in getriggerde pijplijnen kan één create_auto_cdc_from_snapshot_flow() flow een doel delen met één of meer AUTO CDC stromen die in Python of SQL zijn gedefinieerd. De volgende vereisten zijn van toepassing:

Geef elke AUTO CDC flow een unieke naam.
Gebruik hetzelfde aantal sleutels in dezelfde volgorde voor alle stromen. Snapshot-flow sleutelnamen worden zonder hoofdletters vergeleken met AUTO CDC sleutelnamen. Meerdere stromen AUTO CDC moeten identieke sleutelnamen en behuizing gebruiken.
Gebruik exact hetzelfde datatype voor de snapshotversie en de sequencing-kolom van elke AUTO CDC flow.
Definieer geen verwachtingen over de AUTO CDC FROM SNAPSHOT flow.
Maak geen gebruik van IGNORE NULL UPDATES. In Python moet je niet , ignore_null_updates_column_list, of ignore_null_updates_except_column_list op de create_auto_cdc_flow() flows zettenignore_null_updates.
Als je een AUTO CDC flow toevoegt aan een bestaand AUTO CDC FROM SNAPSHOT doelwit, mag het doel geen gebruikerskolommen bevatten waarvan de namen conflicteren met gereserveerde AUTO CDC systeemkolommen.
Gebruik dit patroon niet voor SCD Type 2 of bitemporele doelen.

Zie Voor een voorbeeld, Voeg een backfill toe aan een AUTO CDC SCD Type 1 tabel.

`source` Het argument implementeren

De create_auto_cdc_from_snapshot_flow() functie bevat het source argument. Voor het verwerken van historische momentopnamen wordt verwacht dat het source argument een Python-lambda-functie is die twee waarden retourneert aan de create_auto_cdc_from_snapshot_flow() functie: een Python DataFrame met de momentopnamegegevens die moeten worden verwerkt en een momentopnameversie.

De eerste aanroep van de functie moet een snapshot DataFrame en versie teruggeven. Als de eerste aanroep terugkeert None, mislukt de pipeline-update. Nadat ten minste één snapshot is verwerkt, geef je aan None dat er geen extra snapshots beschikbaar zijn.

Hier volgt de handtekening van de lambda-functie:

lambda Any => Optional[(DataFrame, Any)]

Het argument voor de lambda-functie is de laatst verwerkte momentopnameversie.
De retourwaarde van de lambda-functie is None of een tuple van twee waarden: de eerste waarde van de tuple is een DataFrame dat de momentopname bevat die moet worden verwerkt. De tweede waarde van de tuple is de momentopnameversie die de logische volgorde van de momentopname aangeeft.

Een voorbeeld waarmee de lambda-functie wordt geïmplementeerd en aangeroepen:

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

De Runtime van Lakeflow-pijplijnen voert de volgende stappen uit telkens wanneer de pijplijn die de create_auto_cdc_from_snapshot_flow() functie bevat, wordt geactiveerd:

Hiermee wordt de next_snapshot_and_version functie uitgevoerd om het volgende dataframe voor momentopnamen en de bijbehorende momentopnameversie te laden.
Als de eerste aanroep geen DataFrame teruggeeft, faalt de update. Als een latere aanroep geen DataFrame teruggeeft, wordt de run beëindigd en wordt de pipeline-update als voltooid gemarkeerd.
Detecteert de wijzigingen in de nieuwe momentopname en past deze incrementeel toe op de doeltabel.
Keert terug naar stap 1 om de volgende momentopname en de bijbehorende versie te laden.

Feedback

Is deze pagina nuttig?

Laatst bijgewerkt om 2026-10-05