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......流动（管道）。

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

参数

flow_name

要创建的流的名称。
评论

流的可选说明。
自动 CDC INTO

定义流程的语句是使用 AUTO CDC ... INTO 和 create_auto_cdc_flow_spec。必须包括 AUTO CDC ... INTO 语句或 INSERT INTO 语句。当源查询使用更改数据语义时使用 AUTO CDC ... INTO 。

有关详细信息，请参阅 AUTO CDC INTO （管道）。
自动CDC......摘自快照

一个 AUTO CDC ... INTO 通过比较快照而非读取变更源来推导变更的语句。当源端未启用变更数据采集且仅提供完整快照时，请使用此表单。源代码分为两部分：一个读取快照数据的必 FROM SNAPSHOT (snapshot_query) 填子句，以及选择下一个快照版本的可选 WITH VERSION (version_query) 子句。请参阅 AUTO CDC FROM SNAPSHOT 的工作原理。
- 摘自快照（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 运行时 18.2 及以上版本。

定义该流程为一个 REPLACE USING 流程，替换目标表中所有匹配指定键列的行，其他行保持不动。当你的源是按列键的部分快照系列时使用 REPLACE USING 。 SEQUENCE BY 更新顺序使得密钥的最高序列获胜，即使更新顺序不一致。

至少指定一个键列，且精确指定一 SEQUENCE BY 列。查询必须是流式查询，且 BY NAME 是必需的。 REPLACE USING 不能与 ONCEAUTO CDC ... INTO或结合。

更多信息请参见 “部分快照替换”使用流“。
替换 WHERE 条件

将该流程定义为一个 REPLACE WHERE 流，重新计算并覆盖目标表的目标子集。每次更新时，目标表匹配 condition 中的所有行都会被删除，针对同一谓词范围重新计算源查询，并插入结果。不匹配 condition 的行则保持原样。你不需要在源查询中添加谓词;流水线引擎在读取源代码时会自动应用它。

REPLACE WHERE 使用批处理语义，因此源查询不必是流式查询。 BY NAME 必需。 REPLACE WHERE 不能与 ONCE、 REPLACE USING或 AUTO CDC ... INTO组合使用。

更多信息请参见 “REPLACE WHERE flows”的批处理和独立流表的“REPLACE WHERE flows”。
一次

可选地将流定义为一次性流，例如回填流。通过两种方式使用 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 （管道）

Syntax

参数

例子

反馈

其他资源