Interfejsy API AUTO CDC upraszczają przechwytywanie danych zmian za pomocą potoków.

Potoki Lakeflow upraszczają przechwytywanie zmian danych (CDC) za pomocą interfejsów API AUTO CDC i AUTO CDC FROM SNAPSHOT. Interfejsy API automatyzują złożoność przetwarzania wolno zmieniających się wymiarów (SCD) typu 1 i typu 2 z pliku CDC lub migawek bazy danych. Interfejs AUTO CDC API obsługuje również śledzenie bitemporalne, które rejestruje zmiany w dwóch wymiarach czasowych (beta). Aby dowiedzieć się więcej o SCD typu 1 i typu 2, zobacz Przechwytywanie zmian danych i migawki. Aby uzyskać więcej informacji na temat śledzenia bitemporalnego, zobacz Bitemporal AUTO CDC.

Używany interfejs API zależy od źródła danych zmian:

• AUTO CDC: użyj tej opcji, gdy źródłowa baza danych ma włączony kanał CDC. AUTO CDC przetwarza zmiany ze strumienia danych o zmianach (CDF). Jest obsługiwany zarówno w interfejsach potoku SQL, jak i Python.
• AUTO CDC FROM SNAPSHOT: użyj tej opcji, gdy usługa CDC nie jest włączona w źródłowej bazie danych i dostępne są tylko migawki. Ten interfejs API porównuje migawki w celu określenia zmian, a następnie ich przetwarzania. Jest obsługiwany zarówno w interfejsach potoku SQL, jak i Python.

W przypadku obiektów docelowych SCD typu 1 można także połączyć przepływ AUTO CDC FROM SNAPSHOT z jednym lub kilkoma przepływami AUTO CDC. Stosuj ten wzorzec, gdy źródło zapewnia okresowe, autorytatywne migawki oraz strumień CDC o niższym opóźnieniu między snapshotami.

Oba interfejsy API obsługują aktualizowanie tabel przy użyciu typu SCD 1 i typu 2:

• Użyj typu SCD 1, aby bezpośrednio aktualizować rekordy. Historia nie jest zachowywana dla zaktualizowanych rekordów.
• Użyj typu SCD 2, aby zachować historię rekordów dla wszystkich aktualizacji lub aktualizacji określonego zestawu kolumn.

Tylko w przypadku AUTO CDC można również użyć pamięci dwuczasowej, która rozszerza historię SCD Type 2, aby śledzić zmiany w dwóch wymiarach czasu: czasie biznesowym i czasie systemowym. Bitemporal jest w fazie Beta. Zobacz Bitemporal AUTO CDC.

AUTO CDC Obsługuje również częściowe aktualizacje, w których rekord zmiany aktualizuje tylko podzbiór kolumn. Zobacz Stosowanie aktualizacji częściowych.

API AUTO CDC nie są obsługiwane przez deklaratywne potoki Apache Spark.

Aby uzyskać informacje o składni i innych odwołaniach, zobacz AUTO CDC INTO (pipelines), create_auto_cdc_flow i create_auto_cdc_from_snapshot_flow.

Requirements

Aby korzystać z interfejsów API CDC, pipeline musi być skonfigurowany do używania bezserwerowych potoków Lakeflow lub edycji usługi Lakeflow Pipelines Pro lub Advanced.

Jak działa AUTO CDC

Aby wykonać przetwarzanie CDC z użyciem AUTO CDC, utwórz tabelę strumieniową, a następnie użyj instrukcji AUTO CDC ... INTO w języku SQL lub funkcji create_auto_cdc_flow() w języku Python, aby określić źródło, klucze i sekwencjonowanie dla strumienia zmian. Aby uzyskać wyjaśnienie sposobu działania sekwencjonowania i logiki SCD, zobacz Zmienianie przechwytywania i migawek danych. Zobacz przykłady usługi AUTO CDC.

Dla inicjalnego zasilenia ze źródła ze strumieniem zmian użyj AUTO CDC z przepływem once, a następnie kontynuuj przetwarzanie strumienia zmian. Zobacz Replikowanie zewnętrznej tabeli RDBMS przy użyciu usługi AUTO CDC.

Aby uzyskać szczegółowe informacje o składni, zobacz AUTO CDC INTO (potoki) lub create_auto_cdc_flow.

Jak działa AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT określa zmiany w danych źródłowych, porównując migawki w porządku chronologicznym. Migawki można odczytywać bezpośrednio z tabeli Delta, z plików w magazynie chmurowym lub z JDBC. Jest obsługiwany zarówno w interfejsach potoku SQL, jak i Python.

Aby wykonać przetwarzanie CDC z AUTO CDC FROM SNAPSHOT, stwórz tabelę strumieniową, a następnie zdefiniuj przepływ:

• W SQL używamy formy AUTO CDC ... FROM SNAPSHOT polecenia CREATE FLOW lub klauzuli osadzonej FLOW AUTO CDC w CREATE STREAMING TABLE. Źródło to klauzula wymagana FROM SNAPSHOT (snapshot_query) oraz opcjonalna WITH VERSION (version_query) , która wybiera kolejną wersję snapshotu do przetworzenia. Zobacz CREATE FLOW (pipelines).
• W Pythonie użyj funkcji create_auto_cdc_from_snapshot_flow() do określenia snapshota, kluczy i innych argumentów. Zobacz create_auto_cdc_from_snapshot_flow.

Aby uzyskać szczegółowe informacje na temat dwóch wzorców przystosowania i kiedy ich używać, zobacz Wzorce przetwarzania migawek. Zobacz przykłady AUTO CDC FROM SNAPSHOT.

Połącz AUTO CDC ZE SNAPSHOT i AUTO CDC

Wersja migawkowa oraz kolumna sekwencjonowania CDC tworzą jedną domenę porządkową. Zdarzenie CDC nowsze niż najnowszy snapshot ma pierwszeństwo przed snapshotem. Nowszy snapshot ma pierwszeństwo przed starszymi wydarzeniami CDC i traktuje klucze, które nie występują w snapshotie, jako usunięte w wersji snapshot. Nowsze wydarzenia CDC mogą zachować lub przywrócić te klucze.

Zunifikowany cel musi spełniać następujące wymagania:

• Obiekt docelowy wykorzystuje SCD Type 1.
• Cel ma dokładnie jeden AUTO CDC FROM SNAPSHOT przepływ i jeden lub więcej AUTO CDC przepływów.
• Każdy AUTO CDC flow ma unikalną nazwę.
• Wszystkie przepływy używają tej samej liczby kluczy w tej samej kolejności. Nazwy kluczy snapshot-flow są porównywane bez rozróżniania wielkości liter z nazwami kluczy AUTO CDC. Wiele przepływów AUTO CDC musi używać identycznych nazw kluczy i wielkości liter.
• Wersja migawkowa oraz każda kolumna sekwencjonowania CDC mają dokładnie ten sam typ danych.
• Przepływ AUTO CDC FROM SNAPSHOT nie definiuje oczekiwań.
• Przepływy AUTO CDC nie używają IGNORE NULL UPDATES ani jego wariantów typu listy kolumn.
• Potok przetwarzania korzysta z trybu wyzwalanego.
• Jeśli dodasz AUTO CDC flow do istniejącego AUTO CDC FROM SNAPSHOT targetu, target nie może zawierać kolumn użytkownika, których nazwy kolidują z zastrzeżonymi kolumnami systemowymi AUTO CDC.

Oba AUTO CDC flowy mogą AUTO CDC FROM SNAPSHOT korzystać z interfejsu potoku Python lub SQL. Możesz mieszać przepływy SQL i Python w tym samym obiekcie docelowym.

Użyj once=True w Pythonie lub ONCE w SQL w przepływie AUTO CDC FROM SNAPSHOT, aby raz uzupełnić snapshot, podczas gdy przepływy AUTO CDC nadal przetwarzają nowe zdarzenia przy kolejnych aktualizacjach. Pełne odświeżenie obiektu docelowego powtarza jednorazowy proces snapshot. Dla pełnego przykładu zobacz Dodaj uzupełnienie do tabeli typu 1 AUTO CDC SCD.

Ten wzór nie obsługuje celów SCD typu 2 ani bitemporalnych.

Używanie wielu kolumn do sekwencjonowania

Aby sekwencjonować według wielu kolumn (na przykład znacznik czasu i identyfikator przerwania więzi), użyj elementu , STRUCT aby je połączyć. Interfejs API najpierw sortuje według pierwszego pola, a w przypadku remisu uwzględnia drugie pole i tak dalej.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

Przykłady usługi AUTO CDC

W poniższych przykładach pokazano przetwarzanie SCD typu 1 i 2 przy użyciu źródła strumienia danych zmian. Przykładowe dane tworzą nowe rekordy użytkownika, usuwają rekordy użytkownika i aktualizują rekordy użytkowników. W przykładzie SCD Type 1 ostatnie UPDATE operacje docierają późno i są usuwane z tabeli docelowej, demonstrując obsługę zdarzeń w nieprawidłowej kolejności.

Poniżej przedstawiono rekordy wejściowe używane w tych przykładach. Te dane są tworzone przez uruchomienie zapytania w sekcji Tworzenie przykładowych danych .

Jeśli anulujesz komentarz w ostatnim wierszu w zapytaniu generowania danych przykładowych, wstawia on następujący rekord, który określa obcięcie tabeli (wyczyść tabelę) w pliku sequenceNum=3:

Tworzenie przykładowych danych

Uruchom następujące instrukcje, aby utworzyć przykładowy zestaw danych. Ten kod nie jest przeznaczony do uruchomienia w ramach definicji potoku. Uruchom to z folderu eksploracji w twoim potoku danych, zamiast z folderu przekształceń.

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

Przetwarzanie aktualizacji typu SCD 1

Typ SCD 1 przechowuje tylko najnowszą wersję każdego rekordu. Poniższy przykład odczytuje z powyższego kanału danych zmian i stosuje zmiany do docelowej tabeli przesyłania strumieniowego. Potrzebujesz pipeline, aby uruchomić ten kod. Zobacz Tworzenie i debugowanie potoków ETL za pomocą Edytora Potoków Lakeflow.

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;

Po uruchomieniu przykładu typu SCD 1 tabela docelowa zawiera następujące rekordy:

Użytkownik 123 (Isabel) został usunięty i nie jest wyświetlany. Użytkownik 125 (Mercedes) pokazuje tylko najnowsze miasto (Guadalajara), ponieważ typ SCD 1 zastępuje poprzednie wartości. Wcześniejsza wartość w UPDATE i sequenceNum=5 została porzucona, ponieważ pojawiła się późniejsza aktualizacja w sequenceNum=6.

Po uruchomieniu przykładu z odkomentowanym rekordem TRUNCATE tabela zostanie wyczyszczona pod adresem sequenceNum=3. Oznacza to, że rekordy 124 i 126 nie znajdują się w tabeli, a końcowa tabela docelowa zawiera tylko następujący rekord:

Przetwarzanie aktualizacji typu SCD 2

Typ SCD 2 zachowuje pełną historię zmian, tworząc nowe wiersze dla każdej wersji rekordu z kolumnami __START_AT i __END_AT wskazującymi, kiedy każda wersja była aktywna.

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;

Po uruchomieniu przykładu typu SCD 2 tabela docelowa zawiera następujące rekordy:

Tabela zachowuje pełną historię danych. Użytkownik 123 ma dwie wersje (zakończone sekwencją 6 po usunięciu). Użytkownik 125 ma trzy wersje pokazujące zmiany miasta. Rekordy z __END_AT = null są obecnie aktywne.

Śledzenie podzestawu kolumn przy użyciu typu SCD 2

Domyślnie typ SCD 2 tworzy nową wersję za każdym razem, gdy każda wartość kolumny ulegnie zmianie. Możesz określić podzbiór kolumn do śledzenia, aby zmiany w innych kolumnach aktualizowały bieżącą wersję zamiast generowania nowego rekordu historii.

Poniższy przykład wyklucza kolumnę city ze śledzenia historii:

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)

Ponieważ city zmiany nie są śledzone, aktualizacje miejskie zastępują obecny wiersz zamiast tworzyć nową wersję. Tabela docelowa zawiera następujące rekordy:

Przykłady automatycznego CDC z migawki

W poniższych sekcjach przedstawiono przykłady użycia AUTO CDC FROM SNAPSHOT do przetwarzania obrazów danych w tabele docelowe typu SCD 1 lub SCD 2. Aby uzyskać informacje na temat tego, kiedy używać tego interfejsu API, zobacz Zmienianie przechwytywania i migawek danych.

Przykłady w tej sekcji to implementacje Python i SQL. Aby uzyskać pełną składnię SQL, zobacz CREATE FLOW (pipelines).

Przykład: Przetwarzanie okresowych migawek

Stosuj ten sposób, gdy migawki pojawiają się regularnie i w kolejności. Interfejs Python korzysta z kolejności aktualizacji pipeline’u do wersjonowania. Dla SQL przykład używa osobnej tabeli wersji, aby zidentyfikować, kiedy dostępny jest nowy snapshot.

Migawki można odczytywać z wielu typów źródeł, w tym z tabel Delta, plików w chmurze i połączeń JDBC.

Krok 1. Tworzenie przykładowych danych

Utwórz tabelę zawierającą dane migawki. Uruchom następujący kod z notatnika lub usługi Databricks SQL w folderze explorations Twojego potoku:

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

Krok 2. Uruchom automatyczny CDC z migawki

W tym etapie potrzebujesz pipeline, aby uruchomić kod. Zobacz Tworzenie i debugowanie potoków ETL za pomocą Edytora Potoków Lakeflow.

W Python wybierz typ źródłowy dla widoku snapshot (przykładowy kod tworzenia generuje tabelę Delta):

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

Następnie dodaj tabelę docelową i flow. Implementacja SQL wykorzystuje tabelę snapshot_version do identyfikacji, kiedy dostępny jest nowy snapshot:

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;

Po pierwszym uruchomieniu procesu potokowego wszystkie rekordy są wstawiane jako aktywne zapisy.

Krok 3: Zasymuluj nową migawkę i uruchom ponownie

Zaktualizuj tabelę źródłową, aby zasymulować nadejście nowej migawki (uruchom ten kod z notatnika lub pliku SQL w folderze explorations twojego 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;

Uruchom ponownie potok. AUTO CDC FROM SNAPSHOT Porównuje nową migawkę z poprzednią i wykrywa, że użytkownik 1 został usunięty, użytkownicy 2 i 3 zostały zaktualizowane, a użytkownicy 4 i 6 zostały wstawione. To generuje strumień zmian, i używa AUTO CDC do utworzenia tabeli wyjściowej.

Po drugim uruchomieniu z typem SCD 2 tabela docelowa zawiera następujące rekordy:

Użytkownik 1 został zakończony (usunięty). Użytkownicy 2 i 3 mają dwie wersje pokazujące zmiany w mieście. Nowo wstawiono użytkowników 4 i 6.

Po drugim uruchomieniu z typem SCD 1 tabela docelowa zawiera tylko bieżący stan:

Przykład: przetwarzanie migawek przy użyciu funkcji wersji

Użyj tej metody, jeśli potrzebujesz jawnej kontroli nad kolejnością migawek. Na przykład użyj tego podejścia, gdy wiele migawek przychodzi w tym samym czasie, lub gdy migawki przychodzą w nieprawidłowej kolejności. Definiujesz, jak wybrać następny snapshot i jego numer wersji. Interfejs API przetwarza migawki w kolejności rosnącej wersji:

• Jeśli w pamięci znajduje się wiele migawek, wybierak przykładów przetwarza dostępne migawki w kolejności wersji.
• Jeśli migawka pojawi się w nieodpowiedniej kolejności (na przykład snapshot_3 pojawi się po snapshot_4), selektor przykładów ją wyklucza, ponieważ jej wersja nie jest większa niż ostatnio zatwierdzona wersja.
• Jeśli nie ma nowych migawków, funkcja wersji lub zapytanie nie zwraca wyniku i nie następuje żadne przetwarzanie.

Krok 1. Przygotowywanie plików migawek

Utwórz pliki CSV zawierające dane migawek i dodaj je do lokalizacji zasobu danych lub przechowywania w chmurze. Nazwij pliki chronologicznie (na przykład snapshot_1.csv, snapshot_2.csv).

Każdy plik powinien zawierać kolumny dla userId i city. Przykład:

snapshot_1.csv:

snapshot_2.csv:

Krok 2: Uruchom AUTO CDC FROM SNAPSHOT z funkcją wersji.

W potoku utworz nowy plik Python lub SQL w folderze transformations i dodaj następujący kod. Następnie uruchom potok. Zobacz Tworzenie i debugowanie potoków ETL za pomocą Edytora Potoków Lakeflow.

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;

Po przetworzeniu snapshot_1.csvtabela docelowa zawiera następujące rekordy:

Po przetworzeniu snapshot_2.csvtabela docelowa zawiera następujące rekordy:

Krok 3: Dodaj nowe migawki

Dodaj nowy plik CSV do miejsca przechowywania z danymi, które zostały zmodyfikowane (na przykład ze zmienionymi wartościami miast, nowymi wierszami lub usuniętymi wierszami). Następnie ponownie uruchom potok, aby przetworzyć nową migawkę.

Ograniczenia

• Kolumna sekwencjonowania musi być sortowalnym typem danych. NULL wartości sekwencjonowania nie są obsługiwane.
• Aby strumieniować dane z obiektu docelowego procesu AUTO CDC, odczytuj je z jego strumienia zmian. Aby uzyskać szczegółowe informacje, zobacz Odczytywanie zestawienia danych zmian z tabeli docelowej usługi AUTO CDC.

Dodatkowe zasoby

• Przechwytywanie danych i migawki: zapoznaj się z koncepcjami CDC, migawkami oraz typami SCD.
• Replikuj zewnętrzną tabelę RDBMS za pomocą AUTO CDC: Dowiedz się, jak przeprowadzić początkowe załadowanie danych w przepływie once, a potem kontynuować przetwarzanie zmian.
• Uzupełnianie danych historycznych za pomocą potoków: Dowiedz się, jak uzupełniać cele typu append-only oraz AUTO CDC SCD Type 1.
• Zaawansowane tematy usługi AUTO CDC: Dowiedz się więcej o operacjach zmian w obiektach docelowych usługi AUTO CDC, odczytywaniu źródeł danych zmian i przetwarzaniu metryk.
• Cofnij i odgraj pipeline: Dowiedz się, jak przywrócić cel AUTO CDC do wcześniejszego punktu czasu i przetworzyć zmiany.
• Samouczek: zbudować potok ETL przy użyciu wychwytywania zmian danych