CREATE FLOW (pipeline)

Gunakan pernyataan CREATE FLOW untuk membuat alur atau pengisian ulang bagi tabel dalam sebuah pipeline.

Note

CREATE FLOW Targeting tabel streaming mendukung keduanya AUTO CDC ... INTO dan REPLACE WHERE aliran. Tabel terkelola yang dibuat dengan CREATE TABLE ... FLOW tidak mendukung penangkapan data perubahan: AUTO CDC ... INTO aliran terhadap tabel terkelola gagal dengan MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. CREATE FLOW ke dalam tabel streaming tidak tunduk pada batasan itu. Lihat CREATE TABLE ... FLOW (pipa).

Syntax

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

Parameter-parameternya

  • flow_name

    Nama alur yang akan dibuat.

  • KOMENTAR

    Deskripsi opsional untuk alur.

  • OTOMATIS MENJADI CDC

    Perintah AUTO CDC ... INTO yang menentukan alur, dengan create_auto_cdc_flow_spec. Anda harus menyertakan AUTO CDC ... INTO pernyataan, atau INSERT INTO pernyataan. Gunakan AUTO CDC ... INTO saat kueri sumber menggunakan semantik data perubahan.

    Untuk informasi selengkapnya, lihat AUTO CDC INTO (pipeline).

  • AUTO CDC ... DARI SNAPSHOT

    Pernyataan AUTO CDC ... INTO yang menurunkan perubahan dengan membandingkan snapshot daripada membaca feed perubahan. Gunakan formulir ini ketika penangkapan data perubahan tidak diaktifkan pada sumber dan hanya snapshot lengkap yang tersedia. Sumber ditentukan dalam dua bagian: klausa wajib FROM SNAPSHOT (snapshot_query) yang membaca data snapshot, dan klausa opsional WITH VERSION (version_query) yang memilih versi snapshot berikutnya untuk diproses. Lihat Cara kerja AUTO CDC FROM SNAPSHOT.

    • DARI SNAPSHOT (snapshot_query)

      Required. Sebuah kueri yang membaca data snapshot untuk versi yang dipilih oleh WITH VERSION (...). Mesin membedakan hasil terhadap snapshot yang sudah dikomit sebelumnya untuk menurunkan sisipan, pembaruan, dan penghapusan, lalu menggabungkannya ke target menggunakan KEYS identitas baris dan STORED AS menentukan bagaimana perubahan disimpan.

      Panggil current_snapshot_version() di dalam kueri ini untuk merujuk versi yang dipilih oleh WITH VERSION (...). Jika WITH VERSION (...) tidak ditentukan, current_snapshot_version() tidak dapat dipanggil di dalam FROM SNAPSHOT (...).

      Ketika WITH VERSION (...) dihilangkan, mesin membaca sumber langsung melalui FROM SNAPSHOT (...), dan kueri snapshot hanya berjalan selama pemuatan awal, sementara target tidak memiliki data yang dikomitmenkan dan tidak ada status snapshot yang dikomitmen. Pada pembaruan berikutnya, ketika target sudah berisi data atau memiliki status snapshot yang dikomitmen, alur gagal dengan AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Untuk memproses snapshot di berbagai pembaruan, gunakan WITH VERSION (...).

    • DENGAN VERSI (version_query)

      Optional. Sebuah kueri yang memilih versi snapshot berikutnya untuk diproses. Harus mengembalikan tepat satu kolom tipe yang dapat diurutkan dan baris 0 atau 1. Ketika mengembalikan 1 baris, nilainya harus non-null. Kolom dapat berupa nilai skalar, seperti a BIGINT, atau a STRUCT yang semua bidangnya dapat diurutkan. Kueri versi yang mengembalikan lebih dari satu kolom, lebih dari satu baris, atau nilai null gagal mengalir dengan INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Selama satu pembaruan pipeline, mesin mengulangi langkah-langkah berikut: mengevaluasi kueri versi; jika kueri mengembalikan 0 baris, ia menghentikan pemrosesan alur ini untuk pembaruan saat ini; jika kueri mengembalikan 1 baris, mesin akan menampilkan nilai tersebut melalui current_snapshot_version(), mengevaluasi kueri snapshot, mengkomit snapshot yang dihasilkan, dan mengekspos versi yang telah dikomit melalui last_snapshot_version(). Mesin kemudian mengevaluasi ulang kueri versi untuk memilih versi berikutnya. Satu pembaruan pipeline memproses versi secara berurutan hingga kueri versi tidak mengembalikan baris apa pun.

      Setiap versi yang dikembalikan setelah commit berhasil harus lebih besar dari versi yang sebelumnya komitmen; versi yang tidak meningkat gagal memperbarui dengan APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Tipe data nilai versi harus tetap tidak berubah di seluruh snapshot commit; perubahan tipe data gagal dalam pembaruan dengan AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Penyegaran penuh membersihkan status versi yang tersimpan.

    • KUNCI

      Required. Kolom kunci utama yang digunakan untuk mengidentifikasi baris di seluruh snapshot untuk deteksi perubahan.

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

      Optional. Menentukan bagaimana perubahan disimpan dalam tabel target. Defaultnya adalah SCD TYPE 1.

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

      Optional. Hanya berlaku dengan SCD TYPE 2. Menentukan kolom mana yang memicu baris riwayat baru saat berubah. Sediakan daftar kolom eksplisit atau * EXCEPT (col_list) melacak setiap kolom kecuali yang tercantum.

    Snapshot CDC tidak mendukung WHERE atau SEQUENCE BY. Pengurutan cross-snapshot diekspresikan melalui WITH VERSION (...).

  • target_table

    Tabel yang akan diperbarui. Ini harus berupa tabel streaming.

  • INSERT KE

    Menentukan suatu kueri tabel yang disisipkan ke dalam tabel target. ONCE Jika opsi tidak disediakan, kueri harus berupa kueri streaming. Gunakan kata kunci STREAM untuk menggunakan semantik streaming untuk membaca dari sumbernya. Jika pembacaan mengalami perubahan atau penghapusan pada rekaman yang ada, akan menghasilkan kesalahan. Paling aman untuk membaca dari sumber statis atau yang hanya bisa ditambahkan. Untuk memasukkan data yang memiliki komit perubahan, Anda dapat menggunakan Python dan opsi skipChangeCommits untuk menangani kesalahan.

    INSERT INTO saling eksklusif dengan AUTO CDC ... INTO. Gunakan AUTO CDC ... INTO saat data sumber menyertakan fungsionalitas change data capture (CDC). Gunakan INSERT INTO ketika sumber tidak melakukannya.

    Untuk informasi selengkapnya tentang data streaming, lihat Mengubah data dengan alur.

  • GANTI MENGGUNAKAN ( column_name [, ...] ) URUTAN OLEH sequence_column

    Important

    Fitur ini ada di Beta. Memerlukan Databricks Runtime 18.2 ke atas.

    Mendefinisikan alur sebagai REPLACE USING aliran, yang menggantikan semua baris dalam tabel target yang sesuai dengan kolom kunci yang ditentukan dan membiarkan semua baris lainnya tidak disentuh. Gunakan REPLACE USING ketika sumber Anda adalah serangkaian snapshot parsial yang dikunci berdasarkan kolom. SEQUENCE BY mengatur pembaruan sehingga urutan tertinggi untuk sebuah kunci menang, bahkan ketika pembaruan datang tidak sesuai urutan.

    Tentukan setidaknya satu kolom kunci dan tepat satu SEQUENCE BY kolom. Kueri harus berupa kueri streaming, dan BY NAME wajib dilakukan. REPLACE USING tidak dapat dikombinasikan dengan ONCE atau dengan AUTO CDC ... INTO.

    Untuk informasi lebih lanjut, lihat Penggantian snapshot parsial dengan alur GANTI MENGGUNAKAN.

  • Kondisi GANTI WHERE

    Mendefinisikan aliran sebagai REPLACE WHERE aliran, yang menghitung ulang dan menimpa subset target dari tabel target. Pada setiap pembaruan, semua baris dalam pencocokan condition tabel target dihapus, kueri sumber dihitung ulang untuk rentang predikat yang sama, dan hasilnya dimasukkan. Baris yang tidak cocok condition dibiarkan tidak tersentuh. Anda tidak perlu menambahkan predikat ke kueri sumber; mesin pipeline menerapkannya secara otomatis saat membaca dari sumber.

    REPLACE WHERE menggunakan semantik batch, jadi query sumber tidak harus berupa query streaming. BY NAME diperlukan. REPLACE WHERE tidak dapat dikombinasikan dengan ONCE, REPLACE USING, atau AUTO CDC ... INTO.

    Untuk informasi lebih lanjut, lihat Pemrosesan batch dengan alur REPLACE WHERE dan alur REPLACE WHERE untuk tabel streaming mandiri.

  • SEKALI

    Secara opsional tentukan alur sebagai aliran satu kali, seperti isi ulang. Dengan menggunakan ONCE, alurnya dapat diubah dengan dua cara:

    • Sumber query atau create_auto_cdc_flow_spec bukan tabel streaming.
    • Alur dijalankan satu kali secara default. Jika alur diperbarui dengan pembaruan lengkap, maka ONCE alur akan berjalan kembali untuk membuat ulang data.

    ONCE tidak dapat digunakan dengan REPLACE USING, yang memerlukan sumber streaming.

Examples

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

Untuk informasi lebih lanjut tentang menggabungkan pengisian satu kali dengan CDC yang sedang berlangsung pada target yang sama, lihat Pengisian data historis dengan pipeline.

  • Terakhir diperbarui pada