CREATE FLOW (potoki danych)

Użyj instrukcji CREATE FLOW, aby utworzyć przepływy lub uzupełnienia dla tabel w potoku.

Składnia

CREATE FLOW flow_name [COMMENT comment] AS
{
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_flow_spec |
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_from_snapshot_spec |
  INSERT [ONCE] INTO target_table BY NAME [ replace_using_spec | REPLACE WHERE condition ] query
}

create_auto_cdc_from_snapshot_spec
  FROM SNAPSHOT ( snapshot_query )
  [ WITH VERSION ( version_query ) ]
  KEYS ( key [, ...] )
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 } ]
  [ TRACK HISTORY ON { col_list | * EXCEPT ( col_list ) } ]

replace_using_spec
  REPLACE USING ( column_name [, ...] ) SEQUENCE BY sequence_column

Parametry

• flow_name

  Nazwa przepływu do utworzenia.

• KOMENTARZ

  Opcjonalny opis przepływu.

• AUTO CDC DO

  Instrukcja AUTO CDC ... INTO definiująca przepływ z elementem create_auto_cdc_flow_spec. Musisz dołączyć instrukcję AUTO CDC ... INTO lub instrukcję INSERT INTO . Użyj AUTO CDC ... INTO polecenia , gdy zapytanie źródłowe używa semantyki zmian danych.

  Aby uzyskać więcej informacji, zobacz AUTO CDC INTO (potoki).

• AUTO CDC ... Z SNAPSHOT

  Stwierdzenie AUTO CDC ... INTO , które wyprowadza zmiany przez porównanie migawek zamiast odczytu feedu zmian. Używaj tego formularza, gdy przechwytywanie danych zmian nie jest włączone na źródle i dostępne są tylko pełne migawki. Źródło jest określone w dwóch częściach: wymaganej klauzuli FROM SNAPSHOT (snapshot_query) odczytującej dane snapshota oraz opcjonalnej klauzuli WITH VERSION (version_query) wybierającej kolejną wersję snapshota do przetworzenia. Zobacz Jak działa auto CDC FROM SNAPSHOT.

  • Z SNAPSHOT (snapshot_query)

    Required. Zapytanie, które odczytuje dane migawki dla wersji wybranej przez WITH VERSION (...). Silnik różnica wynik względem wcześniej zatwierdzonego snapshota, aby wyprowadzić wstawki, aktualizacje i usunięcia, a następnie scala je z celem, używając KEYS identyfikacji wiersza oraz STORED AS określenia, jak są przechowywane zmiany.

    Wołaj current_snapshot_version() wewnątrz tego zapytania, aby odwołać się do wersji wybranej przez WITH VERSION (...). Jeśli WITH VERSION (...) nie jest określona, current_snapshot_version() nie jest wywoływalna wewnątrz FROM SNAPSHOT (...).

    Gdy WITH VERSION (...) zostanie pominięty, silnik odczytuje źródło bezpośrednio przez FROM SNAPSHOT (...), a zapytanie snapshot jest wykonywane tylko podczas początkowego ładowania, podczas gdy cel nie ma zadeklarowanych danych ani zadeklarowanego stanu snapshotu. W każdej późniejszej aktualizacji, gdy cel już zawiera dane lub ma zatwierdzony stan migawki, przepływ kończy się z .AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION Aby przetwarzać migawki w wielu aktualizacjach, użyj WITH VERSION (...).

  • Z WERSJĄ (version_query)

    Optional. Zapytanie, które wybiera kolejną wersję snapshotu do przetworzenia. Musi zwracać dokładnie jedną kolumnę typu uporządkowalnego oraz 0 lub 1 wiersz. Gdy zwraca 1 wiersz, wartość ta musi być niezerowa. Kolumna może mieć wartość skalarną, na przykład , BIGINTlub , STRUCT której pola są uporządkowalne. Zapytanie o wersje, które zwraca więcej niż jedną kolumnę, więcej niż jeden wiersz lub wartość null, nie przechodzi przepływu z .INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY

    Podczas jednej aktualizacji potoku silnik powtarza następujące kroki: ocenia zapytanie wersyjne; jeśli zapytanie zwraca 0 wierszy, przestaje przetwarzać ten przepływ dla bieżącej aktualizacji; jeśli zapytanie zwraca 1 wiersz, silnik ujawnia tę wartość przez current_snapshot_version(), ocenia zapytanie snapshot, zatwierdza powstały snapshot i udostępnia zadeklarowaną wersję przez last_snapshot_version(). Silnik następnie ponownie ocenia zapytanie o wersję, aby wybrać następną wersję. Pojedyncza aktualizacja potoku przetwarza wersje w kolejności, aż zapytanie o wersje nie zwróci żadnych wierszy.

    Każda wersja zwracana po udanym commitcie musi być większa niż wcześniej zatwierdzona wersja; wersja niezwiększająca się nie przechodzi aktualizacji z .APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION Typ danych wartości wersji musi pozostać niezmieniony przy zatwierdzaniu snapshotów; zmiana typu danych powoduje niepowodzenie aktualizacji z AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Pełne odświeżenie usuwa stan wersji utrwalonej.

  • KLUCZE

    Required. Główne kolumny klucza służą do identyfikacji wierszy w migawkach w celu wykrywania zmian.

  • STORED AS { SCD TYPE 1 | SCD TYPE 2 }

    Optional. Określa, jak zmiany są przechowywane w docelowej tabeli. Wartość domyślna to SCD TYPE 1.

  • TRACK HISTORY ON { col_list | * EXCEPT (col_list) }

    Optional. Dotyczy tylko z SCD TYPE 2. Określa, które kolumny wywołują nowy wiersz historii po zmianie. Zapewnij wyraźną listę kolumn lub * EXCEPT (col_list) śledzić każdą kolumnę oprócz wymienionych.

  Snapshot CDC nie obsługuje WHERE ani SEQUENCE BY. Kolejność migawek krzyżowych wyraża się przez .WITH VERSION (...)

• target_table

  Tabela do zaktualizowania. Musi to być tabela przesyłania strumieniowego.

• INSERT DO

  Definiuje zapytanie tabeli, które jest wstawiane do tabeli docelowej. ONCE Jeśli opcja nie jest podana, zapytanie musi być zapytaniem przesyłanym strumieniowo. Użyj słowa kluczowego STREAM, aby stosować semantykę przesyłania strumieniowego do odczytu ze źródła. Jeśli odczyt napotka zmianę lub usunięcie istniejącego rekordu, zostanie zgłoszony błąd. Odczyt ze źródeł statycznych lub dołączanych jest najbezpieczniejszy. Aby pozyskiwać dane, które mają zatwierdzenia zmian, możesz użyć Pythona i opcji skipChangeCommits do obsługi błędów.

  INSERT INTO wzajemnie się wyklucza z AUTO CDC ... INTO. Użyj AUTO CDC ... INTO , gdy dane źródłowe obejmują funkcje przechwytywania zmian danych (CDC). Użyj INSERT INTO, gdy źródło tego nie robi.

  Aby uzyskać więcej informacji na temat danych przesyłanych strumieniowo, zobacz Przekształcanie danych za pomocą potoków.

• ZAMIEŃ UŻYWAJĄC ( column_name [, ...] ) SEKWENCJA WEDŁUG sequence_column

  Important

  Ta funkcja jest dostępna w wersji beta. Wymaga Databricks Runtime 18.2 i nowszych.

  Definiuje przepływ jako REPLACE USING flow, który zastępuje wszystkie wiersze w docelowej tabeli odpowiadające określonym kolumnom klucza i pozostawia pozostałe wiersze niezmienione. Używaj, REPLACE USING gdy źródło to seria częściowych migawek przypisanych kolumnom. SEQUENCE BY Zleca aktualizacje tak, aby wygrała najwyższa sekwencja klucza, nawet gdy aktualizacje pojawiają się w nieodpowiedniej kolejności.

  Określ co najmniej jedną kolumnę klucza i dokładnie jedną SEQUENCE BY kolumnę. Zapytanie musi być zapytaniem strumieniowym i BY NAME jest wymagane. REPLACE USING Nie można łączyć z ONCE ani z AUTO CDC ... INTO.

  Więcej informacji można znaleźć w artykule Częściowa wymiana snapshotów za pomocą przepływów REPLACE USING (ZASTENIA).

• Warunek REPLACE WHERE

  Definiuje przepływ jako REPLACE WHERE przepływ, który przetwarza i nadpisuje docelowy podzbiór docelowej tabeli. Przy każdej aktualizacji wszystkie wiersze w docelowej condition tabeli są usuwane, zapytanie źródłowe jest ponownie obliczane dla tego samego zakresu predykatów i wstawiane są wyniki. Niepasujące rzędy condition pozostają nietknięte. Nie musisz dodawać predykatu do zapytania źródłowego; silnik pipeline stosuje go automatycznie podczas odczytu ze źródła.

  REPLACE WHERE Korzysta z semantyki wsadowej, więc zapytanie źródłowe nie musi być zapytaniem strumieniowym. Ciąg BY NAME jest wymagany. REPLACE WHERE Nie można łączyć z ONCE, REPLACE USING, ani AUTO CDC ... INTO.

  Więcej informacji można znaleźć w artykule Przetwarzanie wsadowe z przepływami REPLACE WHERE oraz przepływy REPLACE WHERE dla samodzielnych tabel strumieniowych.

• RAZ

  Opcjonalnie zdefiniuj przepływ jako przepływ jednorazowy, taki jak wypełnienie wsteczne. Używanie ONCE zmienia przepływ na dwa sposoby:

  • Źródło query lub create_auto_cdc_flow_spec nie jest tabelą przesyłania strumieniowego.
  • Domyślnie przepływ jest uruchamiany jeden raz. W przypadku zaktualizowania pipeline'u przez pełne odświeżenie, przepływ ONCE zostanie uruchomiony ponownie w celu odtworzenia danych.

  ONCE nie można używać z REPLACE USING, które wymaga źródła strumieniowego.

Przykłady

-- EXAMPLE 1:
-- Create a streaming table, and add two flows that append data to it:
CREATE OR REFRESH STREAMING TABLE users;

-- first flow into target_table:
CREATE FLOW users_flow AS
INSERT INTO users BY NAME
SELECT * FROM stream(raw_data.users);

-- second flow into target_table:
CREATE FLOW backfill_users AS
INSERT ONCE INTO users BY NAME
SELECT * FROM user_backfill_table;

-- EXAMPLE 2:
-- Create a streaming table, and add a flow that applies CDC changes to it:
CREATE OR REFRESH STREAMING TABLE admins_cdc_target_table;

-- first flow into target_table:
CREATE FLOW admin_cdc_flow AS
AUTO CDC INTO admins_cdc_target_table
FROM stream(cdc_data.admins)
KEYS (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum)
STORED AS SCD TYPE 2;

-- EXAMPLE 3:
-- Create a streaming table, and add a REPLACE USING flow that keeps the latest
-- row for each payment_id from a stream of partial snapshots:
CREATE OR REFRESH STREAMING TABLE payments_latest;

CREATE FLOW payments_replace_flow AS
INSERT INTO payments_latest BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

-- EXAMPLE 4:
-- AUTO CDC FROM SNAPSHOT without WITH VERSION: a one-time initial load from a snapshot table.
-- To process later snapshots on each update, add WITH VERSION (see EXAMPLE 5).
CREATE STREAMING TABLE users (user_id INT, name STRING, email STRING);

CREATE FLOW users_snapshot_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (SELECT * FROM catalog.schema.users_snapshot)
KEYS (user_id)
STORED AS SCD TYPE 1;

-- EXAMPLE 5:
-- AUTO CDC FROM SNAPSHOT with WITH VERSION: pick the next file, then read it as the snapshot:
CREATE STREAMING TABLE orders (order_id INT, product STRING, quantity INT, order_date DATE);

CREATE FLOW orders_cdc AS
AUTO CDC INTO orders
FROM SNAPSHOT (
  SELECT order_id, product, quantity, order_date
  FROM read_files('/Volumes/catalog/schema/landing/orders/', format => 'json')
  WHERE _metadata.file_path = (SELECT version.path FROM current_snapshot_version())
)
WITH VERSION (
  SELECT struct(modification_time, path) AS version
  FROM list_files('/Volumes/catalog/schema/landing/orders/')
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR struct(modification_time, path) > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY modification_time, path
  LIMIT 1
)
KEYS (order_id)
STORED AS SCD TYPE 2;

-- EXAMPLE 6:
-- One-time snapshot backfill plus a streaming CDC flow into the same target.
-- The backfill omits WITH VERSION, so it uses an implicit timestamp version. The
-- streaming flow's SEQUENCE BY column (event_ts) must be a TIMESTAMP so its type
-- matches that implicit version on the shared target.
CREATE STREAMING TABLE customers (
  customer_id INT, name STRING, email STRING, address STRING, event_ts TIMESTAMP
);

CREATE FLOW customers_snapshot_backfill AS
AUTO CDC ONCE INTO customers
FROM SNAPSHOT (SELECT * FROM catalog.schema.customers_snapshot)
KEYS (customer_id)
STORED AS SCD TYPE 1;

CREATE FLOW customers_cdc AS
AUTO CDC INTO customers
FROM STREAM(customers_cdc_events)
KEYS (customer_id)
SEQUENCE BY event_ts
STORED AS SCD TYPE 1;

-- EXAMPLE 7:
-- Create a streaming table, then add a REPLACE WHERE flow that recomputes and
-- overwrites a targeted window of the target table on each update:
CREATE STREAMING TABLE payments_latest;

CREATE FLOW payments_latest AS
INSERT INTO payments_latest BY NAME
REPLACE WHERE payment_date >= date_add(current_date(), -7)
SELECT payment_id, booking_id, status, payment_date
FROM samples.wanderbricks.payments;

Więcej informacji o połączeniu jednorazowego uzupełnienia z trwającym CDC na tym samym celu można znaleźć w artykule Uzupełnianie danych historycznych rurociągami.