CREATE FLOW (işlem hatları)

CREATE FLOW ifadesini bir işlem hattındaki tablolar için akışlar veya geri yüklemeler oluşturmak için kullanın.

Note

CREATE FLOW Bir akış tablosunu hedeflemek hem AUTO CDC ... INTO de REPLACE WHERE akışları destekler. Yönetilen bir tablo değiştirilen CREATE TABLE ... FLOW veri yakalamayı desteklemez: yönetilen bir tabloya karşı bir AUTO CDC ... INTO akış ile MANAGED_TABLE_DOES_NOT_SUPPORT_CDCbaşarısız olur. CREATE FLOW Bir yayın tablosuna bu sınırlamaya tabi değildir. Bakınız CREATE TABLE ... FLOW (boru hatları).

Sözdizimi

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

Parametreler

  • flow_name

    Oluşturulacak akışın adı.

  • YORUM

    Akış için isteğe bağlı bir açıklama.

  • OTOMATIK CDC INTO

    AUTO CDC ... INTO ile akışı tanımlayan bir create_auto_cdc_flow_spec ifadesi. Ya bir AUTO CDC ... INTO ifadesi ya da bir INSERT INTO ifadesi eklemelisiniz. Kaynak sorgu değişiklik verisi semantiğini kullandığında kullanın AUTO CDC ... INTO .

    Daha fazla bilgi için bkz. AUTO CDC INTO (işlem hatları).

  • OTO CDC ... ANLIKTAN'DAN

    AUTO CDC ... INTO Değişim akışını okumak yerine anlık görüntüleri karşılaştırarak değişiklikleri çıkaran bir ifade. Kaynak üzerinde değişiklik veri yakalama etkin olmadığında ve sadece tam anlık görüntüler mevcut olduğunda bu formu kullanın. Kaynak iki bölümde belirtilir: anlık veriyi okuyan zorunlu FROM SNAPSHOT (snapshot_query) bir madde ve işlenecek bir sonraki anlık versiyonu seçen isteğe WITH VERSION (version_query) bağlı bir cümle. Bkz. ANLıK GÖRÜNTÜDEN OTOMATIK CDC nasıl çalışır?

    • ANLIKTAN (snapshot_query)

      Required. Seçilen WITH VERSION (...)versiyon için anlık verisini okuyan bir sorgu. Motor, sonucu önceden belirlenmiş anlık görüntüyle karşılaştırarak eklemeleri, güncellemeleri ve silmeleri türetir ve bunları hedef ile birleştirerek KEYS satır kimliği ve STORED AS değişikliklerin nasıl saklanacağını belirler.

      Bu sorgu içinde current_snapshot_version() ile seçilen WITH VERSION (...)sürümü referans almak için çağırın. Belirtilmemişse WITH VERSION (...) , current_snapshot_version() içinde FROM SNAPSHOT (...)çağrılamaz .

      Atladığında WITH VERSION (...) , motor kaynağı doğrudan , FROM SNAPSHOT (...)üzerinden okur ve anlık soru yalnızca ilk yükleme sırasında çalışır, hedefin ise taahhüt edilmiş veri ve bağlanmış anlık görüntü durumu yoktur. Daha sonraki herhangi bir güncellemede, hedef zaten veri içerdiğinde veya kararlı bir anlık duruma sahipse, akış ile AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSIONbaşarısız olur. Birden fazla güncellemede anlık görüntüleri işlemek için WITH VERSION (...).

    • VERSIYON (version_query)

      Optional. Bir sonraki anlık versiyonu işleyen bir sorgu. Tam olarak bir sütun ve sıralanabilir türde bir sütun ve ya 0 ya da 1 satır döndürmelidir. 1 satır döndürdüğünde, değer sıfır olmayan olmalıdır. Sütun, örneğin a BIGINTgibi skaler bir değer olabilir veya tüm alanları sıralanabilir olan a STRUCT olabilir. Birden fazla sütun, birden fazla satır veya null değer döndüren bir sürüm sorgusu, akışı başarısız olur.INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY

      Bir boru hattı güncellemesi sırasında, motor aşağıdaki adımları tekrarlar: sürüm sorgusunu değerlendirir; sorgu 0 satır dönerse, mevcut güncelleme için bu akışı işlemeyi durdurur; sorgu 1 satır dönerse, motor bu değeri current_snapshot_version() üzerinden açıklar, anlık soruyu değerlendirir, ortaya çıkan anlık görüntüyü uygular ve last_snapshot_version() üzerinden belirlenmiş versiyonu açıklar. Motor, sonraki sürümü seçmek için sürüm sorgusunu yeniden değerlendirir. Tek bir boru hattı güncellemesi, sürümleri sırayla işliyor ve sürüm sorgusu satır döndürmez.

      Başarılı bir taahhüt sonrası geri dönen her versiyon, önceden taahhüt edilen sürümden büyük olmalıdır; artmayan bir sürüm, güncellemede başarısız olur.APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION Sürüm değerinin veri tipi, snapshot commit'ler arasında değişmemelidir; veri tipi değişikliği, güncellemede başarısız olur ve AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Tam yenileme devam eden sürüm durumunu temizler.

    • ANAHTARLAR

      Required. Değişim algısı için anlık görüntüler boyunca satırları tanımlamak için kullanılan ana anahtar sütunlar.

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

      Optional. Değişikliklerin hedef tabloda nasıl saklanacağını belirtir. Varsayılan değer: SCD TYPE 1.

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

      Optional. Sadece ile SCD TYPE 2geçerlidir. Hangi sütunların değiştiğinde yeni bir geçmiş satırını tetiklediğini belirtir. Ya açık bir sütun listesi sağlayın ya da * EXCEPT (col_list) listelenenler hariç her sütunu takip etmek için.

    Snapshot CDC WHERE desteklemiyor veya SEQUENCE BY. Çapraz anlık sıralaması ile WITH VERSION (...)ifade edilir.

  • target_table

    Güncelleştirilecek tablo. Bu bir Akış tablosu olmalıdır.

  • INSERT İÇİNE

    Hedef tabloya eklenen tablo sorgusunu tanımlar. Seçenek ONCE, verilmezse sorgu bir akış sorgusu olmalıdır. Kaynaktan okumak üzere akış semantiğini kullanmak için STREAM anahtar sözcüğünü kullanın. Okuma işlemi var olan bir kayıtta bir değişiklik veya silme işlemiyle karşılaşırsa bir hata oluşur. Statik veya yalnızca ekleme kaynaklarından okumak en güvenlidir. Değişiklik taahhütleri içeren verileri içeri çekmek için Python'ı ve skipChangeCommits seçeneğini hataları işlemek için kullanabilirsiniz.

    INSERT INTO ile AUTO CDC ... INTObirbirini dışlar. Kaynak veriler değişiklik verisi yakalama (CDC) işlevselliği içerdiğinde kullanın AUTO CDC ... INTO . Kaynakta kullanılmadığında INSERT INTO kullanın.

    Akış verileri hakkında daha fazla bilgi için bkz. İşlem hatları ile veri dönüştürme.

  • YERINE ÇALIN ( column_name [, ...] ) DIZISI sequence_column

    Important

    Bu özellik Beta sürümündedir. Databricks Runtime 18.2 ve üzeri gerektirir.

    Akışı bir akış REPLACE USING olarak tanımlar; hedef tablodaki tüm satırları belirtilen anahtar sütunlara uydurur ve diğer tüm satırlar dokunulmadan bırakır. Kaynağınız sütunlara göre anahtarlanmış kısmi anlık görüntüler dizisi olduğunda kullanın REPLACE USING . SEQUENCE BY güncellemeleri sipariş eder, böylece anahtar için en yüksek sıra, güncellemeler sırasız gelse bile kazanır.

    En az bir anahtar sütunu ve tam olarak bir SEQUENCE BY sütun belirtin. Sorgu akış sorgusu olmalı ve BY NAME zorunludur. REPLACE USINGile veya ONCEbirleştirilemezAUTO CDC ... INTO.

    Daha fazla bilgi için, REPLACE USING akışlarıyla kısmi anlık görüntü değiştirme bölümünü inceleyebilirsiniz.

  • REPLACE WHERE durumu

    Akışı, hedef tablonun hedef alt kümesini yeniden hesaplayan ve üzerine yazan bir REPLACE WHERE akış olarak tanımlar. Her güncellemede, hedef tablo eşleştirmesindeki condition tüm satırlar silinir, kaynak sorgu aynı önlem aralığı için yeniden hesaplanır ve sonuçlar eklenir. Eşleşmeyen sıralar condition dokunulmadan kalır. Kaynak sorguya önlem eklemenize gerek yok; pipeline motoru kaynaktan okurken otomatik olarak uygular.

    REPLACE WHERE toplu anlam kullanır, bu yüzden kaynak sorgu akış sorgusu olmak zorunda değildir. BY NAME gereklidir. REPLACE WHERE, , veya AUTO CDC ... INTOile birleştirilemez. REPLACE USINGONCE

    Daha fazla bilgi için, bağımsız akış tabloları için REPLACE WHERE akışları ile toplu işleme ve REPLACE WHERE akışları sayfalarına bakınız.

  • BİR DEFA

    İsteğe bağlı olarak akışı bir kerelik akış olarak (örneğin, bir geri doldurma) tanımlayın. Kullanımı ONCE , akışı iki şekilde değiştirir:

    • Kaynak query veya create_auto_cdc_flow_spec bir akış tablosu değil.
    • Akış varsayılan olarak bir kez çalıştırılır. Eğer işlem hattı eksiksiz bir yenilemeyle güncellenirse, ONCE akış verileri yeniden oluşturmak için tekrar çalıştırılır.

    ONCE ile REPLACE USINGkullanılamaz, bu da bir akış kaynağı gerektirir.

Örnekler

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

Aynı hedefte tek seferlik geri doldurmayı devam eden CDC ile birleştirmek hakkında daha fazla bilgi için Geçmiş verileri boru hatlarıyla geri doldurma bölümünü inceleyebilirsiniz.

  • Son güncelleştirme: