CREATE FLOW （管線）

使用陳述 CREATE FLOW 式為管線中的資料表建立流程或回填。

Note

CREATE FLOW 針對串流表支援兩者 AUTO CDC ... INTO 與 REPLACE WHERE 流程。以 CREATE TABLE ... FLOW 建立的受管理資料表不支援變更資料擷取：對受管理資料表的AUTO CDC ... INTO流程在下失敗。MANAGED_TABLE_DOES_NOT_SUPPORT_CDC CREATE FLOW 進入串流表則不受這個限制。參見CREATE TABLE......流（管線）。

語法

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

參數

flow_name

要建立的流程名稱。
評論

流程的可選說明。
自動 CDC 啟用

一個AUTO CDC ... INTO陳述式用來定義流程，並具有create_auto_cdc_flow_spec。您必須包含 AUTO CDC ... INTO 陳述式，或是 INSERT INTO 陳述式。當來源查詢使用變更資料語意時使用 AUTO CDC ... INTO 。

如需詳細資訊，請參閱 AUTO CDC INTO （管線）。
自動疾病控制中心......摘自快照

一個 AUTO CDC ... INTO 透過比較快照而非讀取變更訂閱源來推導變更的陳述式。當來源未啟用變更資料擷取且僅有完整快照時，請使用此表單。來源由兩部分指定：一個讀取快照資料的必要 FROM SNAPSHOT (snapshot_query) 子句，以及選擇下一個要處理的快照版本的可選 WITH VERSION (version_query) 子句。請參考 SNAPSHOT 中的自動 CDC 運作方式。
- 摘自快照（snapshot_query）
  
  Required. 一個查詢，讀取由 WITH VERSION (...)選擇版本的快照資料。引擎會將結果與先前提交的快照進行比對，以推導插入、更新與刪除，並利用資料列身份STORED AS與變更的儲存方式合併至目標KEYS。
  
  在此查詢中呼叫 current_snapshot_version（）以參考由 WITH VERSION (...)選擇的版本。若 WITH VERSION (...) 未指定 current_snapshot_version() ，則在內無法呼叫 FROM SNAPSHOT (...)。
  
  當 WITH VERSION (...) 被省略時，引擎會直接 FROM SNAPSHOT (...)讀取原始碼，快照查詢僅在初始載入時執行，而目標沒有已提交的資料，也沒有已提交的快照狀態。在任何後續更新中，當目標已包含資料或已提交快照狀態時，流程將以為失敗。AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION 若要處理跨多個更新的快照，請使用 WITH VERSION (...)。
- 版本（version_query）
  
  Optional. 一個查詢，選擇下一個要處理的快照版本。它必須回傳恰好一欄可排序型態，且列數為 0 或 1。當回傳 1 列時，該值必須是非空。欄位可以是純量值，例如， BIGINT或 STRUCT 是所有欄位皆可排序的。若版本查詢回傳多於一欄、多列或空值，則流程失敗。INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY
  
  在一次管線更新中，引擎會重複以下步驟：評估版本查詢;若查詢回傳 0 列，則停止處理當前更新的流程;若查詢回傳 1 列，引擎會透過 current_snapshot_version（）暴露該值，評估快照查詢，提交結果快照，並透過 last_snapshot_version（）公開已提交版本。引擎接著會重新評估版本查詢以選擇下一個版本。單一管線更新會依序處理版本，直到版本查詢回傳無列為止。
  
  成功提交後回傳的每個版本必須大於先前提交的版本;非遞增版本則更新失敗。APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION 版本值的資料型別必須在快照提交期間保持不變;資料型別變更會使更新失敗。AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED 完整刷新會清除持久版本狀態。
- 鑰匙
  
  Required. 主要鍵欄位用於識別快照間的列，以便變更偵測。
- STORED AS { SCD TYPE 1 | SCD TYPE 2 }
  
  Optional. 指定變更如何儲存在目標資料表中。預設值為 SCD TYPE 1。
- TRACK HISTORY ON { col_list | * EXCEPT (col_list) }
  
  Optional. 僅 SCD TYPE 2適用於。指定哪些欄位在變更時會觸發新的歷史列。提供明確的欄位列表，或 * EXCEPT (col_list) 追蹤除列出欄位外的所有欄位。
快照 CDC 不支援 WHERE 或 SEQUENCE BY。交叉快照排序透過表示。WITH VERSION (...)
target_table

要更新的資料表。這必須是串流資料表。
INSERT 轉換為

定義插入至目標資料表的資料庫查詢。如果未提供選項，則 ONCE 查詢必須是串流查詢。使用 STREAM 關鍵詞來使用串流語意從來源讀取。如果讀取遇到現有記錄的變更或刪除，則會拋出錯誤。閱讀靜態或只能添加的來源是最安全的。若要匯入包含變更提交的數據，您可以使用 Python 和 skipChangeCommits 選項來處理錯誤。

INSERT INTO 與 AUTO CDC ... INTO互斥。當來源資料包含變更資料擷取（CDC）功能時使用 AUTO CDC ... INTO 。當來源沒有時使用 INSERT INTO 。

如需串流資料的詳細資訊，請參閱用管線轉換資料。
取代使用（ column_name [， ...]）依sequence_column順序排列

Important

這項功能位於測試版 (Beta) 中。需要 Databricks Runtime 18.2 及以上版本。

定義流程為一個 REPLACE USING 流程，會替換目標資料表中所有符合指定鍵欄位的列，並保留其他所有列。當你的來源是一系列以欄位鍵入的部分快照時，請使用 REPLACE USING 。 SEQUENCE BY 指令更新順序，使得金鑰的最高序列勝出，即使更新順序不正確。

至少指定一個金鑰欄位，且精確指定一 SEQUENCE BY 欄。查詢必須是串流查詢，且 BY NAME 是必須的。 REPLACE USING 無法與 ONCEAUTO CDC ... INTO或結合。

欲了解更多資訊，請參閱部分快照替換為 REPLACE USING 流程。
替換 WHERE 條件

定義流程為一個 REPLACE WHERE 流程，重新計算並覆蓋目標資料表的目標子集。每次更新時，目標表中所有匹配 condition 的列都會被刪除，針對同一謂詞範圍重新計算來源查詢，並插入結果。不匹配 condition 的行則保持不變。你不需要把謂詞加入來源查詢;管線引擎會在讀取原始碼時自動套用這個條件。

REPLACE WHERE 使用批次語意，因此來源查詢不一定要是串流查詢。 BY NAME 是必要的。 REPLACE WHERE 不能與 ONCE、 REPLACE USING或 AUTO CDC ... INTO結合使用。

欲了解更多資訊，請參閱使用 REPLACE WHERE 流程的批次處理及獨立串流資料表的 REPLACE WHERE 流程。
ONCE

選擇性地將流程定義為一次性流程，例如回填。使用 ONCE 將以兩種方式改變流程：
- 來源 query 或 create_auto_cdc_flow_spec 不是串流資料表。
- 預設情況下，流程會執行一次。如果管道已更新為完整刷新，則ONCE流程會再度執行以重新創建資料。
ONCE 不能搭配 REPLACE USING，因為需要串流來源。

範例

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

關於在同一目標上將一次性回填與持續的CDC合併，請參見「用管線回填歷史資料」。

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-10-15

CREATE FLOW （管線）

語法

參數

範例

意見反應

其他資源