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

Lakeflow-pipelines förenklar insamling av ändringsdata (CDC) med API:erna AUTO CDC och AUTO CDC FROM SNAPSHOT . Dessa API:er automatiserar komplexiteten i att beräkna långsamt föränderliga dimensioner (SCD) typ 1 och typ 2 från antingen ett CDC-flöde eller databasögonblicksbilder. API:et AUTO CDC stöder även bitmporal spårning, som registrerar ändringar över två tidsdimensioner (Beta). Mer information om SCD-typ 1 och typ 2 finns i Ändra datainsamling och ögonblicksbilder. Mer information om bitemporal tracking finns i Bitemporal AUTO CDC.

Det API som du använder beror på källan för dina ändringsdata:

• AUTO CDC: Använd detta när källdatabasen har ett CDC-flöde aktiverat. AUTO CDC bearbetar ändringar från en ändringsdatafeed (CDF). Det stöds i både SQL- och Python-pipelinegränssnitten.
• AUTO CDC FROM SNAPSHOT: Använd detta när CDC inte är aktiverat i källdatabasen och endast ögonblicksbilder är tillgängliga. Det här API:et jämför ögonblicksbilder för att fastställa ändringar och bearbetar dem sedan. Det stöds i både SQL- och Python-pipelinegränssnitten.

För SCD Typ 1-mål kan du också kombinera ett AUTO CDC FROM SNAPSHOT flöde med ett eller flera AUTO CDC flöden. Använd detta mönster när en källa tillhandahåller periodiska auktoritativa snapshots och ett CDC-flöde med lägre latens mellan snapshots.

Båda API:erna stöder uppdatering av tabeller med scd typ 1 och typ 2:

• Använd SCD Typ 1 för att uppdatera dataposter 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.

Endast för AUTO CDC kan du också använda bitemporal storage, vilket utökar historiken för SCD typ 2 för att spåra ändringar över två tidsdimensioner: affärstid och systemtid. Bitemporal är i betaversion. Se Bitemporal AUTO CDC.

AUTO CDC stöder också partiella uppdateringar, där en ändringspost endast uppdaterar en delmängd kolumner. Se Tillämpa partiella uppdateringar.

AUTO CDC API:er stöds inte av Apache Spark deklarativa pipelines.

Syntax och andra referenser hittar du i AUTO CDC INTO (pipelines), create_auto_cdc_flow och create_auto_cdc_from_snapshot_flow.

Kravspecifikation

Om du vill använda CDC-API:erna måste din pipeline vara konfigurerad för att använda serverlösa Lakeflow-pipelines eller Lakeflow-pipelinesversionerna Pro eller Advancedutgåvorna.

Så här fungerar AUTO CDC

Om du vill utföra CDC-bearbetning med AUTO CDCskapar du 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. En förklaring av hur sekvensering och SCD-logik fungerar finns i Ändra datainsamling och ögonblicksbilder. Se AUTO CDC-exempel.

För initial hydrering från en källa med ett ändringsflöde använder du AUTO CDC med ett once flöde och fortsätter sedan att bearbeta ändringsflödet. Se Replikera en extern RDBMS-tabell med AUTO CDC.

Syntaxinformation finns i AUTO CDC INTO (pipelines) eller create_auto_cdc_flow.

Så här fungerar AUTO CDC FRÅN SNAPSHOT

AUTO CDC FROM SNAPSHOT avgör ändringar i källdata genom att jämföra ögonblicksbilder i ordning. Du kan läsa ögonblicksbilder från en Delta-tabell, molnlagringsfiler eller JDBC direkt. Det stöds i både SQL- och Python-pipelinegränssnitten.

För att utföra CDC-bearbetning med AUTO CDC FROM SNAPSHOT, skapa en strömningstabell och definiera sedan flödet:

• I SQL, använd AUTO CDC ... FROM SNAPSHOT-formen av instruktionen CREATE FLOW eller den inbäddade FLOW AUTO CDC-satsen i CREATE STREAMING TABLE. Källan är en obligatorisk FROM SNAPSHOT (snapshot_query) klausul och en valfri WITH VERSION (version_query) klausul som väljer nästa snapshot-version att bearbeta. Se CREATE FLOW (pipelines).
• I Python, använd create_auto_cdc_from_snapshot_flow() funktionen för att specificera snapshot, nycklar och andra argument. Se create_auto_cdc_from_snapshot_flow.

Mer information om de två inmatningsmönstren och när du ska använda var och en finns i Mönster för bearbetning av ögonblicksbilder. Se exempel på AUTO CDC FROM SNAPSHOT.

Kombinera AUTO CDC FROM SNAPSHOT och AUTO CDC

Snapshot-versionen och CDC-sekvenskolumnen bildar en ordningsdomän. En CDC-händelse som är nyare än den senaste ögonblicksbilden har företräde framför ögonblicksbilden. En nyare snapshot går före äldre CDC-händelser och behandlar nycklar som saknas i snapshoten som raderade vid snapshot-versionen. Nyare CDC-händelser kan bevara eller återställa dessa nycklar.

Ett enhetligt mål måste uppfylla följande krav:

• Målet använder SCD typ 1.
• Målet har exakt ett AUTO CDC FROM SNAPSHOT flöde och ett eller flera AUTO CDC flöden.
• Varje AUTO CDC flöde har ett unikt namn.
• Alla flöden använder samma antal nycklar i samma ordning. Snapshot-flödesnyckelnamn jämförs utan hänsyn till versaler/gemener med AUTO CDC nyckelnamn. Flera AUTO CDC flöden måste använda identiska nyckelnamn och versaler/gemener.
• Snapshot-versionen och varje CDC-sekvenseringskolumn har exakt samma datatyp.
• Flödet AUTO CDC FROM SNAPSHOT definierar inte förväntningar.
• Flödena AUTO CDC använder inte IGNORE NULL UPDATES eller dess kolumnlistvarianter.
• Pipelinen använder utlöst läge.
• Om du lägger till ett AUTO CDC flöde till ett befintligt AUTO CDC FROM SNAPSHOT mål får målet inte innehålla användarkolumner vars namn krockar med reserverade AUTO CDC systemkolumner.

Både AUTO CDC- och AUTO CDC FROM SNAPSHOT-flöden kan använda Python- eller SQL-pipelinegränssnittet. Du kan blanda SQL- och Python-flöden i samma mål.

Använd once=True i Python eller ONCE i SQL på AUTO CDC FROM SNAPSHOT flödet för att fylla i en snapshot en gång medan flödena AUTO CDC fortsätter att bearbeta nya händelser vid efterföljande uppdateringar. En fullständig uppdatering av målobjektet kör om engångsflödet för ögonblicksbilden. För ett komplett exempel, se Lägg till en backfill till en AUTO CDC SCD Typ 1-tabell.

Detta mönster stöder inte SCD typ 2 eller bitemporala mål.

Använda flera kolumner för sekvensering

Om du vill sekvenseras efter flera kolumner (till exempel en tidsstämpel och ett ID för att bryta banden) använder du en STRUCT för att kombinera dem. API:t sorterar efter det första fältet först, och vid lika värden beaktar det det andra fältet och så vidare.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

python

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

AUTO CDC-exempel

I följande exempel visas SCD-typ 1 och typ 2-bearbetning med hjälp av en ändringsdataflödeskälla. Exempeldata skapar nya användarposter, tar bort en användarpost och uppdaterar användarposter. I scd typ 1-exemplet kommer de sista UPDATE åtgärderna sent och tas bort från måltabellen, vilket visar händelsehantering utan ordning.

Följande är de indataposter som används i de här exemplen. Dessa data skapas genom att köra frågan i avsnittet Skapa exempeldata .

Om du avkommenterar den sista raden i exempeldatagenereringsfrågan infogar den följande post som anger att tabellen ska trunkeras (rensa tabellen) på sequenceNum=3:

Skapa exempeldata

Kör följande instruktioner för att skapa en exempeldatauppsättning. Den här koden är inte avsedd att köras som en del av en pipelinedefinition. Kör den från utforskningsmappen för din pipeline i stället för mappen transformationer.

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)
);

Bearbeta SCD Typ 1-uppdateringar

SCD Typ 1 sparar endast den senaste versionen av varje post. Följande exempel läser från ändringsdataflödet som skapades ovan och tillämpar ändringar på en målströmningstabell. Du behöver en pipeline för att köra denna kod. Se Utveckla och felsöka ETL-pipelines med 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;

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

Användare 123 (Isabel) har tagits bort och visas inte. Användare 125 (Mercedes) visar endast den senaste staden (Guadalajara) eftersom SCD Typ 1 skriver över tidigare värden. Det tidigare UPDATE på sequenceNum=5 togs bort för att en senare uppdatering hade kommit på sequenceNum=6.

Efter att ha kört exemplet med posten TRUNCATE okommenterad rensas tabellen på sequenceNum=3. Det innebär att dataposter 124 och 126 inte finns i tabellen, och den slutliga måltabellen endast innehåller följande datapost:

Bearbeta SCD Typ 2-uppdateringar

SCD-typ 2 bevarar en fullständig historik över ändringar genom att skapa nya rader för varje version av en post, med __START_AT och __END_AT kolumner som anger när varje version var aktiv.

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;

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

Tabellen bevarar fullständig historik. Användare 123 har två versioner (avslutades vid sekvens 6 när den togs bort). Användare 125 har tre versioner som visar ändringar i staden. Poster som innehåller __END_AT = null är för närvarande aktiva.

Spåra en kolumnuppsättning med SCD Type 2

Som standard skapar SCD Typ 2 en ny version när ett kolumnvärde ändras. Du kan ange en delmängd av kolumner som ska spåras, så att ändringar i andra kolumner uppdaterar den aktuella versionen på plats i stället för att generera en ny historikpost.

I följande exempel undantas city kolumnen från historikspårning:

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)

Eftersom city ändringarna inte spåras skriver stadsuppdateringar över den aktuella raden i stället för att skapa en ny version. Måltabellen innehåller följande poster:

EXEMPEL PÅ AUTOMATISK CDC FRÅN ÖGONBLICKSBILD

Följande avsnitt innehåller exempel på hur du använder AUTO CDC FROM SNAPSHOT för att bearbeta ögonblicksbilder till SCD-måltabeller av typ 1 eller typ 2. Bakgrund om när du ska använda det här API:et finns i Ändra datainsamling och ögonblicksbilder.

Exemplen i detta avsnitt inkluderar Python- och SQL-implementationer. För komplett SQL-syntax, se CREATE FLOW (pipelines).

Exempel: Bearbeta periodiska ögonblicksbilder

Använd detta tillvägagångssätt när snapshots anländer regelbundet och i ordning. Python-gränssnittet använder pipelineuppdateringsordningen för versionsnumrering. För SQL använder exemplet en separat versionstabell för att identifiera när en ny snapshot är tillgänglig.

Du kan läsa ögonblicksbilder från flera källtyper, inklusive Delta-tabeller, molnlagringsfiler och JDBC-anslutningar.

Steg 1: Skapa exempeldata

Skapa en tabell som innehåller ögonblicksbilddata. Kör följande kod från en notebook-fil eller Databricks SQL i explorations-mappen för din pipeline:

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);

Steg 2: Kör AUTO CDC FRÅN SNAPSHOT

Du behöver en pipeline för att köra koden i detta steg. Se Utveckla och felsöka ETL-pipelines med Lakeflow Pipelines Editor.

För Python, välj en källtyp för snapshotvyn (exempelkoden genererar en Delta-tabell):

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")

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()
  )

Lägg sedan till måltabellen och flödet. SQL-implementeringen använder tabellen snapshot_version för att identifiera när en ny ögonblicksbild är tillgänglig:

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;

Efter den första pipelinekörningen infogas alla poster som aktiva rader.

Steg 3: Simulera en ny ögonblicksbild och kör igen

Uppdatera källtabellen för att simulera en ny snapshot som anländer (kör denna kod från en anteckningsbok eller SQL-fil i explorations mappen i din 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;

Kör pipelinen igen. AUTO CDC FROM SNAPSHOT jämför den nya ögonblicksbilden med den föregående och identifierar att användare 1 togs bort, användare 2 och 3 uppdaterades och användare 4 och 6 infogades. Detta genererar en ändringsfeed och använder AUTO CDC för att skapa utdatatabellen.

Efter den andra körningen med SCD Typ 2 innehåller måltabellen följande poster:

Användare 1 avslutades (togs bort). Användare 2 och 3 har båda två versioner som visar deras stadsändringar. Användare 4 och 6 infogades nyligen.

Efter den andra körningen med SCD Typ 1 visar måltabellen endast det aktuella tillståndet:

Exempel: Bearbeta ögonblicksbilder med hjälp av versionsfunktioner

Använd den här metoden när du behöver explicit kontroll över beställning av ögonblicksbilder. Använd till exempel den här metoden när flera ögonblicksbilder anländer samtidigt eller anländer i fel ordning. Du definierar hur du väljer nästa snapshot och dess versionsnummer. API:et bearbetar ögonblicksbilder i stigande versionsordning:

• Om flera snapshots finns i lagring bearbetar exempelväljaren tillgängliga snapshots i versionsordning.
• Om en snapshot anländer i fel ordning (till exempel snapshot_3 anländer efter snapshot_4), exkluderar exempelväljaren den eftersom dess version inte är större än den senast committerade versionen.
• Om det inte finns några nya snapshots returnerar versionsfunktionen eller frågan inget resultat och ingen bearbetning sker.

Steg 1: Förbereda ögonblicksbildfiler

Skapa CSV-filer som innehåller ögonblicksbildsdata och lägg till dem på en volym- eller molnlagringsplats. Namnge filerna kronologiskt (till exempel snapshot_1.csv, snapshot_2.csv).

Varje fil ska innehålla kolumner för userId och city. Som exempel:

snapshot_1.csv:

snapshot_2.csv:

Steg 2: Kör AUTO CDC FROM SNAPSHOT med en versionsfunktion

I en pipeline, skapa en ny Python- eller SQL-fil i transformations mappen och lägg till följande kod. Kör sedan pipelinen. Se Utveckla och felsöka ETL-pipelines med 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;

Efter bearbetningen snapshot_1.csvinnehåller måltabellen följande poster:

Efter bearbetningen snapshot_2.csvinnehåller måltabellen följande poster:

Steg 3: Lägga till nya ögonblicksbilder

Lägg till en ny CSV-fil till lagringsplatsen med ändrade data (till exempel ändrade stadsvärden, nya rader eller borttagna rader). Kör sedan pipelinen igen för att bearbeta den nya ögonblicksbilden.

Begränsningar

• Sekvenseringskolumnen måste vara en sorterbar datatyp. NULL sekvenseringsvärden stöds inte.
• Om du vill strömma data från målet för en AUTO CDC-process läser du från dess ändringsflöde. Mer information finns i Läs ett ändringsflöde från en AUTO CDC måltabell.

Ytterligare resurser

• Ändra datainsamling och ögonblicksbilder: Lär dig mer om CDC-begrepp, ögonblicksbilder och SCD-typer.
• Replikera en extern RDBMS-tabell med AUTO CDC: Lär dig hur du utför inledande hydrering med ett once flöde och fortsätter sedan att bearbeta ändringar.
• Bakåtfylla historiska data med pipelines: Lär dig hur du bakåtfyller append-only-mål och AUTO CDC SCD Type 1-mål.
• Avancerade auto CDC-ämnen: Lär dig mer om ändringsåtgärder för AUTOMATISKA CDC-mål, läsning av ändringsdataflöden och bearbetningsmått.
• Spola tillbaka och spela upp en pipeline: Lär dig hur du återställer ett AUTO CDC-mål till en tidigare tidpunkt och bearbetar ändringar.
• Självstudie: Skapa en ETL-pipeline med hjälp av insamling av ändringsdata