AUTO CDC FLOW-klausul

Gäller för: Databricks SQL

FLOW AUTO CDC Använd -satsen med CREATE STREAMING TABLE för att bearbeta CDC-poster (Change Data Capture) från en källa till en strömmande tabell.

Tidigare användes instruktionen MERGE INTO ofta för bearbetning av CDC-poster på Azure Databricks. Kan dock MERGE INTO ge felaktiga resultat på grund av poster som inte är sekvenserade eller kräver komplex logik för att ordna om poster.

AUTO CDC förenklar CDC genom att automatiskt hantera out-of-order-poster. Du anger nycklar för att identifiera poster, en sekvenskolumn för beställning och om resultat ska lagras som SCD typ 1 (direktuppdateringar) eller SCD typ 2 (historikspårning).

Syntax

  FLOW AUTO CDC
  FROM source
  KEYS ( keys )
  [ IGNORE NULL UPDATES [ ON { columnList | * EXCEPT ( exceptColumnList ) } ] ]
  [ APPLY AS DELETE WHEN condition ]
  [ APPLY AS TRUNCATE WHEN condition ]
  SEQUENCE BY orderByColumn
  [ SYSTEM SEQUENCE BY systemOrderByColumn ]
  [ COLUMNS { columnList | * EXCEPT ( exceptColumnList ) } ]
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL } ]
  [ TRACK HISTORY ON { columnList | * EXCEPT ( exceptColumnList ) } ]
  [ COLUMNS TO UPDATE columnName ]

Standardbeteendet för INSERT och UPDATE händelser är att uppdatera CDC-händelser från källan: uppdatera alla rader i måltabellen som matchar de angivna nycklarna eller infoga en ny rad när en matchande post inte finns i måltabellen. Hantering av DELETE händelser kan anges med villkoret APPLY AS DELETE WHEN .

Parameters

• source

  Datakällan. Källan måste vara en strömmande källa. Använd nyckelordet STREAM för att använda strömmande semantik för att läsa från källan. Om läsningen påträffar en ändring eller borttagning av en befintlig post utlöses ett fel. Det är säkrast att läsa från statiska eller endast tilläggskällor.

  Mer information om strömmande data finns i Transformera data med pipelines.

• KEYS

  Kolumnen eller kombinationen av kolumner som unikt identifierar en rad i källdata. Värdena i dessa kolumner används för att identifiera vilka CDC-händelser som gäller för specifika poster i måltabellen.

  Om du vill definiera en kombination av kolumner använder du en kommaavgränsad lista med kolumner.

  Den här satsen krävs.

• IGNORE NULL UPDATES

  Tillåter inmatning av uppdateringar som innehåller en delmängd av målkolumnerna. När en CDC-händelse matchar en befintlig rad och IGNORE NULL UPDATES anges behåller kolumner med ett null värde sina befintliga värden i målet. Detta gäller även kapslade kolumner med ett null värde.

  För partiella uppdateringar lägger du till en ON sats för att styra vilka kolumner som ignorerar null värden:

  • IGNORE NULL UPDATES ON columnList: Endast de listade kolumnerna behåller sina befintliga värden när det inkommande värdet är null. Alla andra kolumner använder explicita null värden.
  • IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList): alla kolumner utom de som visas behåller sina befintliga värden när det inkommande värdet är null. Kolumnerna i listan använder explicita null värden.

  Mer information finns i Tillämpa partiella uppdateringar.

  Den här satsen är valfri.

  Standardvärdet är att skriva över befintliga kolumner med null värden.

• APPLY AS DELETE WHEN

  Anger när en CDC-händelse bör behandlas som en DELETE i stället för en uppdatering.

  För SCD-typ 2-källor behålls den borttagna raden tillfälligt som en gravsten i den underliggande Delta-tabellen för att hantera oordnade data, och en vy skapas i metaarkivet som filtrerar bort dessa gravstenar. Kvarhållningsintervallet kan konfigureras med pipelines.cdc.tombstoneGCThresholdInSecondstabellegenskapen.

  Den här satsen är valfri.

• APPLY AS TRUNCATE WHEN

  Anger när en CDC-händelse ska behandlas som en fullständig tabell TRUNCATE. Eftersom den här satsen utlöser en fullständig trunkering av måltabellen bör den endast användas för specifika användningsfall som kräver den här funktionen.

  APPLY AS TRUNCATE WHEN Satsen stöds endast för SCD-typ 1. SCD-typ 2 stöder inte trunceringsoperationen.

  Den här satsen är valfri.

• SEQUENCE BY

  Kolumnnamnet som anger den logiska ordningen för CDC-händelser i källdata. Bearbetningen i pipelinen använder den här sekvenseringen för att hantera ändringshändelser som kommer i oordning.

  Om flera kolumner behövs för sekvensering använder du ett STRUCT uttryck: det sorteras efter det första structfältet först, sedan efter det andra fältet om det finns ett oavgjort resultat och så vidare.

  Angivna kolumner måste vara sorterbara datatyper.

  Den här satsen krävs.

• SYSTEM SEQUENCE BY

  Important

  Bitemporal AUTO CDC är i Beta.

  Kolumnnamnet som anger den systemtid då varje CDC-händelse är känd för systemet. Används med STORED AS BITEMPORAL för att spåra ändringar över både affärstid (SEQUENCE BY) och systemtid. Se Bitemporal AUTO CDC.

  Angivna kolumner måste vara sorterbara datatyper.

  Den här satsen är valfri och gäller endast för bitemporal-tabeller.

• COLUMNS

  Anger en delmängd av kolumner som ska inkluderas i måltabellen. Du kan antingen:

  • Ange den fullständiga listan med kolumner som ska inkluderas: COLUMNS (userId, name, city).
  • Ange en lista med kolumner som ska undantas: COLUMNS * EXCEPT (operation, sequenceNum)

  Den här satsen är valfri.

  Standardvärdet är att inkludera alla kolumner i måltabellen COLUMNS när satsen inte har angetts.

• STORED AS

  Om du vill lagra poster som SCD typ 1, SCD typ 2 eller bitemporal.

  Ange till för att BITEMPORAL spåra ändringar över både affärstid och systemtid. Bitemporal kräver SYSTEM SEQUENCE BY och är i Beta. Se Bitemporal AUTO CDC.

  Den här satsen är valfri.

  Standardvärdet är SCD typ 1.

• TRACK HISTORY ON

  Anger en delmängd av resultatkolumner för att generera historikposter när det sker ändringar i de specificerade kolumnerna. Du kan antingen:

  • Ange den fullständiga listan med kolumner som ska spåras: TRACK HISTORY ON userId, name, city.
  • Ange en lista över kolumner som ska undantas från spårning: TRACK HISTORY ON * EXCEPT (operation, sequenceNum)

  Den här satsen är valfri. Standardvärdet är att spåra historiken för alla utdatakolumner när det finns ändringar, motsvarande TRACK HISTORY ON *.

• COLUMNS TO UPDATE

  Anger namnet på en källkolumn som för varje ändringspost innehåller den uppsättning kolumner som ska uppdateras som en matris med kolumnnamnssträngar (array<string>). Kolumner som inte finns i matrisen behåller sina befintliga målvärden, medan kolumner som visas skrivs från källan, inklusive explicita null värden.

  Använd den här satsen för partiella uppdateringar när varje ändringspost uppdaterar en annan uppsättning kolumner och du måste tillämpa explicita null värden.

  Du kan inte använda COLUMNS TO UPDATE tillsammans med IGNORE NULL UPDATES, och det stöds inte för bitemporal-tabeller.

  Mer information finns i Tillämpa partiella uppdateringar.

  Den här satsen är valfri.

AUTOMATISK CDC FRÅN SNAPSHOT

Använd klausulens FROM SNAPSHOT form FLOW AUTO CDC för att härleda ändringar genom att jämföra snapshots istället för att läsa ett ändringsflöde. Använd detta formulär när ändringsdatafångst inte är aktiverat på källkoden och endast fullständiga ögonblicksbilder är tillgängliga. Källkoden specificeras med en obligatorisk klausul och en valfri FROM SNAPSHOT (snapshot_query)WITH VERSION (version_query) klausul som väljer nästa snapshot-version att bearbeta.

  FLOW AUTO CDC [ ONCE ]
  FROM SNAPSHOT ( snapshot_query )
  [ WITH VERSION ( version_query ) ]
  KEYS ( key [, ...] )
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 } ]
  [ TRACK HISTORY ON { columnList | * EXCEPT ( exceptColumnList ) } ]

• FROM SNAPSHOT krävs och läser snapshot-data för den valda versionen. Motorn diffifierar resultatet mot den tidigare committerade snapshoten för att härleda insättningar, uppdateringar och borttagningar. Anropa current_snapshot_version() i denna fråga för att referera till versionen som valts av WITH VERSION.
• WITH VERSION är valfritt och väljer nästa snapshot-version att bearbeta. Inuti den, ring last_snapshot_version() för att referera till den senaste versionen. En enda pipelineuppdatering bearbetar versioner i ordning tills versionsfrågan inte ger några rader. När WITH VERSION utelämnas körs snapshot-frågan endast under den initiala lasten, medan målet inte har någon committerad data eller snapshot-tillstånd. Vid en senare uppdatering utvärderas ett icke-flödeONCE igen och misslyckas med AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION, medan ett färdigt ONCE flöde hoppas över efter sin första lyckade commit.

Snapshot CDC stöder WHERE inte eller SEQUENCE BY. För fullständiga parameterbeskrivningar, se CREATE FLOW (pipelines) och How AUTO CDC FROM SNAPSHOT fungerar.

Exempel

-- SCD type 1: apply CDC changes with direct updates (no history)
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

-- SCD type 2: retain a history of changes, with delete handling
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2;

-- SCD type 2 with history tracking on specific columns
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  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);

-- SCD type-2 dimension with a generated surrogate key and automatic clustering
> CREATE OR REFRESH STREAMING TABLE ${target_catalog}.silver.dim_customer (
    customer_sk BIGINT GENERATED ALWAYS AS IDENTITY,
    customer_id INT,
    email       STRING,
    segment     STRING,
    city        STRING,
    __START_AT  TIMESTAMP,
    __END_AT    TIMESTAMP
  )
  COMMENT 'SCD Type 2 customer dimension with generated surrogate key'
  CLUSTER BY AUTO
  FLOW AUTO CDC
  FROM stream(${target_catalog}.bronze.customer_changes)
  KEYS (customer_id)
  SEQUENCE BY updated_at
  STORED AS SCD TYPE 2;

-- AUTO CDC FROM SNAPSHOT: one-time initial load of a snapshot table
> CREATE STREAMING TABLE users
  FLOW AUTO CDC ONCE
  FROM SNAPSHOT (SELECT * FROM catalog.schema.users_snapshot)
  KEYS (user_id)
  STORED AS SCD TYPE 1;

Relaterade artiklar

• CREATE STREAMING TABLE
• API:er för AUTOMATISK CDC: Förenkla insamling av ändringsdata med pipelines
• AUTO CDC INTO (rörledningar)
• CREATE FLOW (pipelines)
• last_snapshot_version funktion
• current_snapshot_version funktion
• Flöden