Dela via

API:er för AUTOMATISK CDC: Förenkla insamling av ändringsdata med pipelines

Lakeflow Spark deklarativa pipelines (SDP) förenklar ändringsdatainsamlingen (CDC) med API:erna AUTO CDC och AUTO CDC FROM SNAPSHOT.

Anmärkning

AUTO CDC API:erna ersätter API:erna APPLY CHANGES och har samma syntax. APPLY CHANGES API:erna är fortfarande tillgängliga, men Databricks rekommenderar att du använder API:erna AUTO CDC i deras ställe.

Vilket gränssnitt du använder beror på källan för ändringsdata:

  • Använd AUTO CDC för att bearbeta ändringar från ett ändringsdataflöde (CDF).
  • Använd AUTO CDC FROM SNAPSHOT (offentlig förhandsversion och endast tillgänglig för Python) för att bearbeta ändringar i databasögonblicksbilder.

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 API:et stöds i pipelline SQL- och Python-gränssnitten. AUTO CDC FROM SNAPSHOT API:et stöds i Python-gränssnittet. AUTO CDC API:er stöds inte av Apache Spark deklarativa pipelines.

Både AUTO CDC och AUTO CDC FROM SNAPSHOT stöder uppdatering av tabeller med SCD typ 1 och typ 2:

  • Använd SCD typ 1 för att uppdatera poster direkt. Historik behålls inte för uppdaterade register.
  • Använd SCD typ 2 för att behålla en historik över poster, antingen på alla uppdateringar eller vid uppdateringar av en angiven uppsättning kolumner.

För syntax och andra referenser, se AUTO CDC för pipelines (SQL), AUTO CDC för pipelines (Python) och AUTO CDC FROM SNAPSHOT for pipelines (Python).

Anmärkning

Den här artikeln beskriver hur du uppdaterar tabeller i dina pipelines baserat på ändringar i källdata. Information om hur du registrerar och frågar ändringsinformation på radnivå för Delta-tabeller finns i Använda Delta Lake-ändringsdataflöde i Azure Databricks.

Kravspecifikation

Om du vill använda CDC-API:erna måste din pipeline konfigureras för att använda serverlös SDP eller SDP Pro eller Advancedutgåvor.

Hur implementeras CDC med API:et FÖR AUTOMATISK CDC?

Genom att automatiskt hantera out-of-sequence-poster säkerställer AUTO CDC-API:et korrekt bearbetning av CDC-poster och tar bort behovet av att utveckla komplex logik för hantering av out-of-sequence-poster. Du måste ange en kolumn i källdata för att sekvensera poster, så att API:erna kan tolka det som en kontinuerligt ökande representation av rätt sekvens på källdata. Pipelines hanterar automatiskt data som kommer i fel ordning. För SCD typ 2-ändringar sprider pipelines korrekta sekvenseringsvärden till måltabellens kolumner __START_AT och __END_AT. Det bör finnas en distinkt uppdatering per nyckel vid varje sekvenseringsvärde och NULL-sekvenseringsvärden stöds inte.

Om du vill utföra CDC-bearbetning med AUTO CDCskapar du först en strömningstabell och använder sedan -instruktionen AUTO CDC ... INTOcreate_auto_cdc_flow() i SQL eller funktionen i Python för att ange källa, nycklar och sekvensering för ändringsflödet. För att skapa måluppspelningstabellen, använd CREATE OR REFRESH STREAMING TABLE-instruktionen i SQL eller create_streaming_table()-funktionen i Python. Se exempel på SCD-typ 1 och typ 2-bearbetning .

För information om syntax, se pipelines SQL-referensen eller Python-referensen.

Hur implementeras CDC med API:et AUTO CDC FROM SNAPSHOT ?

Viktigt!

API:et AUTO CDC FROM SNAPSHOT finns i offentlig förhandsversion.

AUTO CDC FROM SNAPSHOT är ett deklarativt API som effektivt avgör ändringar i källdata genom att jämföra en serie ögonblicksbilder i ordning och sedan köra bearbetningen som krävs för CDC-bearbetning av posterna i ögonblicksbilderna. AUTO CDC FROM SNAPSHOT stöds endast av Python-pipelines-gränssnittet.

AUTO CDC FROM SNAPSHOT stöder inmatning av ögonblicksbilder från flera källtyper:

  • Använd periodisk inmatning av ögonblicksbilder för att mata in ögonblicksbilder från en befintlig tabell eller vy. AUTO CDC FROM SNAPSHOT har ett enkelt, strömlinjeformat gränssnitt som stöder regelbunden inmatning av ögonblicksbilder från ett befintligt databasobjekt. En ny ögonblicksbild matas in med varje pipelineuppdatering och inmatningstiden används som ögonblicksbildversion. När en pipeline körs i kontinuerligt läge matas flera ögonblicksbilder in med varje pipelineuppdatering under en period som bestäms av inställningen för utlösarintervall för flödet som innehåller bearbetningen AUTO CDC FROM SNAPSHOT .
  • Använd historisk inmatning av ögonblicksbilder för att bearbeta filer som innehåller ögonblicksbilder av databasen, till exempel ögonblicksbilder som genererats från en Oracle- eller MySQL-databas eller ett informationslager.

Om du vill utföra CDC-bearbetning från valfri källtyp med AUTO CDC FROM SNAPSHOTskapar du först en strömmande tabell och använder create_auto_cdc_from_snapshot_flow() sedan funktionen i Python för att ange ögonblicksbilden, nycklarna och andra argument som krävs för att implementera bearbetningen. Se exemplen på periodisk inmatning av ögonblicksbilder och historisk inmatning av ögonblicksbilder .

Ögonblicksbilderna som skickas till API:et måste vara i stigande ordning efter version. Om SDP identifierar en ögonblicksbild utan ordning utlöses ett fel.

Syntaxinformation finns i python-referensen för pipelines.

Använda flera kolumner för sekvensering

Du kan sekvensera med flera kolumner (till exempel en tidsstämpel och ett ID för att bryta lika resultat), du kan använda en STRUCT för att kombinera dem: det sorterar först efter det första fältet i STRUCT och i händelse av lika resultat beaktas det andra fältet och så vidare.

Exempel i SQL:

SEQUENCE BY STRUCT(timestamp_col, id_col)

Exempel i Python:

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

Begränsningar

Kolumnen som används för sekvensering måste vara en sorterbar datatyp.

Exempel: SCD-typ 1 och SCD typ 2-bearbetning med CDF-källdata

Följande avsnitt innehåller exempel på SCD-typ 1- och typ 2-frågor som uppdaterar måltabeller baserat på källhändelser från ett ändringsdataflöde som:

  1. Skapar nya användarposter.
  2. Tar bort en användarregistrering.
  3. Uppdaterar användarregister. I exemplet SCD typ 1 kommer de sista UPDATE åtgärderna in sent och tas bort från måltabellen, vilket demonstrerar hanteringen av händelser utanför ordningen.

I följande exempel förutsätter vi att du är bekant med att konfigurera och uppdatera pipelines. Se Självstudie: Skapa en ETL-pipeline med hjälp av insamling av ändringsdata.

Om du vill köra de här exemplen måste du börja med att skapa en exempeldatauppsättning. Se Generera testdata.

Följande är indataposterna för dessa exempel:

userId name city Operation sekvensnummer
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lilja Cancun INSERT 2
123 noll noll DELETE 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Om du avkommenterar den sista raden i exempeldata infogas följande post som anger var poster ska trunkeras:

userId name city Operation sekvensnummer
noll noll noll TRUNKERA 3

Anmärkning

Alla följande exempel har alternativ för att ange både DELETE åtgärder och TRUNCATE operationer, men varje är valfri.

Bearbeta SCD-typ 1-uppdateringar

I följande exempel visas bearbetning av SCD typ 1-uppdateringar:

python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW flowname AS AUTO CDC INTO
  target
FROM
  stream(cdc_data.users)
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;

När du har kört SCD typ 1-exemplet innehåller måltabellen följande poster:

userId name city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lilja Cancun

När du har kört SCD typ 1-exemplet med den ytterligare TRUNCATE posten, beskäres poster 124 och 126 på grund av TRUNCATE åtgärden vid sequenceNum=3, och då innehåller måltabellen följande post:

userId name city
125 Mercedes Guadalajara

Bearbeta SCD-typ 2-uppdateringar

I följande exempel visas bearbetning av SCD typ 2-uppdateringar:

python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_flow
AS AUTO CDC INTO
  target
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;

Efter att ha kört SCD typ 2-exemplet innehåller måltabellen följande poster:

userId name city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 noll
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 noll
126 Lilja Cancun 2 noll

En SCD-typ 2-fråga kan också ange en delmängd av utdatakolumner som ska spåras för historik i måltabellen. Ändringar i andra kolumner uppdateras på plats istället för att generera nya historikposter. I följande exempel visas hur du city undantar kolumnen från spårning:

I följande exempel visas hur du använder spårningshistorik med SCD-typ 2:

python

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

@dp.view
def users():
  return spark.readStream.table("cdc_data.users")

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_flow
AS AUTO CDC INTO
  target
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)

När du har kört det här exemplet utan den ytterligare TRUNCATE posten innehåller måltabellen följande poster:

userId name city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 noll
125 Mercedes Guadalajara 2 noll
126 Lilja Cancun 2 noll

Generera testdata

Koden nedan tillhandahålls för att generera en exempeldatauppsättning för användning i exempelfrågorna som finns i den här självstudien. Förutsatt att du har rätt autentiseringsuppgifter för att skapa ett nytt schema och skapa en ny tabell kan du köra dessa instruktioner med antingen en notebook-fil eller Databricks SQL. Följande kod är inte avsedd att köras som en del av en pipeline-definition:

CREATE SCHEMA IF NOT EXISTS cdc_data;

CREATE TABLE
  cdc_data.users
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 above batch at sequenceNum 6 will be the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Exempel: Periodisk bearbetning av ögonblicksbilder

I följande exempel visas SCD-typ 2-bearbetning som matar in ögonblicksbilder av en tabell som lagras på mycatalog.myschema.mytable. Resultatet av bearbetningen skrivs till en tabell med namnet target.

mycatalog.myschema.mytable registerposter med tidsstämpeln 2024-01-01 00:00:00

Key Värde
1 a1
2 a2

mycatalog.myschema.mytable registreringar vid tidsstämpeln 2024-01-01 12:00:00

Key Värde
2 b2
3 a3 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
 return spark.read.table("mycatalog.myschema.mytable")

dp.create_streaming_table("target")

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

Efter att ha bearbetat snapshotarna innehåller måltabellen följande records:

Key Värde __START_AT __END_AT
1 a1 2024-01-01 00:00:00 2024-01-01 12:00:00
2 a2 2024-01-01 00:00:00 2024-01-01 12:00:00
2 b2 2024-01-01 12:00:00 noll
3 a3 2024-01-01 12:00:00 noll

Exempel: Bearbetning av historiska ögonblicksbilder

I följande exempel visas SCD-typ 2-bearbetning som uppdaterar en måltabell baserat på källhändelser från två ögonblicksbilder som lagras i ett molnlagringssystem:

Ögonblicksbild vid timestamp, lagrad i /<PATH>/filename1.csv

Key Spårningskolumn Ej Spårningskolumn
1 a1 b1
2 a2 b2
4 a4 b4

Ögonblicksbild vid timestamp + 5, lagrad i /<PATH>/filename2.csv

Key Spårningskolumn Ej Spårningskolumn
2 a2_new b2
3 a3 b3
4 a4 b4_new

I följande kodexempel visas bearbetning av SCD typ 2-uppdateringar med dessa ögonblicksbilder:

from pyspark import pipelines as dp

def exist(file_name):
  # Storage system-dependent function that returns true if file_name exists, false otherwise

# This function returns a tuple, where the first value is a DataFrame containing the snapshot
# records to process, and the second value is the snapshot version representing the logical
# order of the snapshot.
# Returns None if no snapshot exists.
def next_snapshot_and_version(latest_snapshot_version):
  latest_snapshot_version = latest_snapshot_version or 0
  next_version = latest_snapshot_version + 1
  file_name = "dir_path/filename_" + next_version + ".csv"
  if (exist(file_name)):
    return (spark.read.load(file_name), next_version)
   else:
     # No snapshot available
     return None

dp.create_streaming_live_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = next_snapshot_and_version,
  keys = ["Key"],
  stored_as_scd_type = 2,
  track_history_column_list = ["TrackingCol"]
)

Efter att ha bearbetat snapshotarna innehåller måltabellen följande records:

Key Spårningskolumn Ej Spårningskolumn __START_AT __END_AT
1 a1 b1 1 2
2 a2 b2 1 2
2 a2_new b2 2 noll
3 a3 b3 2 noll
4 a4 b4_new 1 noll

Lägga till, ändra eller ta bort data i en direktuppspelningstabell

Om din pipeline publicerar tabeller till Unity Catalog kan du använda DML-instruktioner ( datamanipuleringsspråk ), inklusive infognings-, uppdaterings-, borttagnings- och sammanslagningsinstruktioner, för att ändra målströmningstabeller som skapats av AUTO CDC ... INTO -instruktioner.

Anmärkning

  • DML-instruktioner som ändrar tabellschemat för en strömmande tabell stöds inte. Se till att DML-uttrycken inte försöker utveckla tabellschemat.
  • DML-instruktioner som uppdaterar en strömmande tabell kan endast köras i ett delat Unity Catalog-kluster eller ett SQL-lager med Databricks Runtime 13.3 LTS och senare.
  • Eftersom direktuppspelning kräver tilläggsdatakällor anger du flaggan skipChangeCommits när du läser källströmningstabellen om bearbetningen kräver strömning från en källströmningstabell med ändringar (till exempel av DML-instruktioner). När skipChangeCommits anges ignoreras transaktioner som tar bort eller ändrar poster i källtabellen. Om din bearbetning inte kräver en strömmande tabell kan du använda en materialiserad vy (som inte har append-only-begränsningen) som måltabell.

Eftersom Lakeflow Spark Deklarativa pipelines använder en angiven SEQUENCE BY kolumn och sprider lämpliga sekvenseringsvärden till kolumnerna __START_AT och __END_AT i måltabellen (för SCD-typ 2), måste du se till att DML-uttryck använder giltiga värden för dessa kolumner för att upprätthålla rätt ordning på poster. Se Hur implementeras CDC med API:et FÖR AUTOMATISK CDC?.

Mer information om hur du använder DML-instruktioner med strömmande tabeller finns i Lägga till, ändra eller ta bort data i en strömmande tabell.

I följande exempel infogas en aktiv post med en startsekvens på 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Läsa ett ändringsdataflöde från en AUTO CDC-måltabell

I Databricks Runtime 15.2 och senare kan du läsa ett ändringsdataflöde från en strömmande tabell som utgör måltavlan för AUTO CDC eller AUTO CDC FROM SNAPSHOT querys på samma sätt som du läser ett ändringsdataflöde från andra Delta-tabeller. Följande krävs för att läsa ändringsdataflödet från en målströmningstabell:

  • Målsströmningstabellen måste publiceras till Unity Catalog. Se Använd Unity Catalog med pipelines.
  • Om du vill läsa ändringsdataflödet från målströmningstabellen måste du använda Databricks Runtime 15.2 eller senare. Om du vill läsa ändringsdataflödet i en annan pipeline måste pipelinen konfigureras för att använda Databricks Runtime 15.2 eller senare.

Du läser ändringsdataflödet från en målströmningstabell som skapades i Lakeflow Spark deklarativa pipelines på samma sätt som att läsa ett ändringsdataflöde från andra Delta-tabeller. Mer information om hur du använder funktionen Delta-ändringsdataflöde, inklusive exempel i Python och SQL, finns i Använd Delta Lake-ändringsdataflöde i Azure Databricks.

Anmärkning

Posten för ändringsdataflöde innehåller metadata som identifierar typen av ändringshändelse. När en post uppdateras i en tabell, innehåller medatadatan för associerade ändringsposter vanligtvis värden som är inställda på _change_type och update_preimage händelser.

Värdena skiljer sig dock _change_type om uppdateringar görs i måluppspelningstabellen som inkluderar ändring av primärnyckelvärden. När ändringar inkluderar uppdateringar av primära nycklar anges metadatafälten _change_type till insert och delete händelser. Ändringar i primära nycklar kan ske när manuella uppdateringar görs i ett av nyckelfälten med en UPDATE eller MERGE instruktion eller, för SCD-tabeller av typ 2, när fältet __start_at ändras för att återspegla ett tidigare startsekvensvärde.

Frågan AUTO CDC avgör de primära nyckelvärdena, som skiljer sig åt för SCD-typ 1- och SCD-typ 2-bearbetning:

  • För SCD-typ 1-bearbetning och Python-gränssnittet för pipelines är den primära nyckeln värdet för keys-parametern i create_auto_cdc_flow()-funktionen. För SQL-gränssnittet är den primära nyckeln de kolumner som definieras av KEYS -satsen i -instruktionen AUTO CDC ... INTO .
  • För SCD typ 2 är den primära nyckeln parametern keys eller KEYS -satsen plus returvärdet från coalesce(__START_AT, __END_AT) åtgärden, där __START_AT och __END_AT är motsvarande kolumner från måluppspelningstabellen.

Hämta data om poster som bearbetas av en CDC-fråga i pipelines

Anmärkning

Följande mått registreras endast av AUTO CDC frågor och inte av AUTO CDC FROM SNAPSHOT frågor.

Följande mått samlas in av AUTO CDC frågor:

  • num_upserted_rows: Antalet utdatarader som har infogats till datamängden under en uppdatering.
  • num_deleted_rows: Antalet befintliga utdatarader som tagits bort från datauppsättningen under en uppdatering.

Måttet num_output_rows, som är utdata för icke-CDC-flöden, samlas inte in för AUTO CDC sökfrågor.

Vilka dataobjekt används för CDC-bearbetning i en pipeline?

Anmärkning

  • Dessa datastrukturer gäller endast för AUTO CDC bearbetning, inte AUTO CDC FROM SNAPSHOT bearbetning.
  • Dessa datastrukturer gäller endast när måltabellen publiceras till Hive-metaarkivet. Om en pipeline publiceras i Unity Catalog är de interna bakgrundstabellerna inte tillgängliga för användarna.

När du deklarerar måltabellen i Hive-metaarkivet skapas två datastrukturer:

  • En vy med det namn som tilldelats måltabellen.
  • En intern stödtabell som används av pipelinen för att hantera CDC-bearbetning. Den här tabellen namnges genom att lägga till __apply_changes_storage_ framför måltabellens namn.

Om du till exempel deklarerar en måltabell med namnet dp_cdc_targetvisas en vy med namnet dp_cdc_target och en tabell med namnet __apply_changes_storage_dp_cdc_target i metaarkivet. Genom att skapa en vy kan Lakeflow Spark Declarative Pipelines filtrera bort den extra informationen, till exempel gravstenar och versioner, som krävs för att hantera data som inte är i ordning. För att visa de bearbetade uppgifterna, använd mål-vy. Eftersom schemat för __apply_changes_storage_ tabellen kan ändras för att stödja framtida funktioner eller förbättringar bör du inte fråga tabellen om produktionsanvändning. Om du lägger till data manuellt i tabellen antas posterna komma före andra ändringar eftersom versionskolumnerna saknas.

Ytterligare resurser

  • Last updated on