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 (パイプライン)」を参照してください。
自動CDC ...スナップショットより

変更フィードを読むのではなく、スナップショットを比較して変更を導き出す AUTO CDC ... INTO ステートメントです。ソースで変更データキャプチャが有効になっておらず、完全なスナップショットのみが利用可能の場合は、このフォームを使用してください。ソースは2つの部分で指定されます。スナップショットデータを読み取る必須の FROM SNAPSHOT (snapshot_query) 節と、次に処理するスナップショットのバージョンを選択するオプションの WITH VERSION (version_query) 節です。スナップショットからの自動 CDC のしくみを参照してください。
- スナップショット(snapshot_query年)より
  
  Required. WITH VERSION (...)が選択したバージョンのスナップショットデータを読み取るクエリです。エンジンは、結果を以前にコミットしたスナップショットと比較して挿入、更新、削除を導き出し、 KEYS for row identityと STORED AS を使ってターゲットにマージして変更の保存方法を決定します。
  
  このクエリ内で 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. 次に処理するスナップショットバージョンを選択するクエリです。順序付け可能なタイプの列を正確に1つ、行数は0または1でなければなりません。 1行を返す場合、その値はnullでなければなりません。列は BIGINTのようなスカラー値であったり、すべてのフィールドが順序付けられる STRUCT であったりします。複数の列、複数の行、またはnull値を返すバージョンクエリは、 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

この機能はベータ版です。 Databricks Runtime 18.2以上が必要です。

フローを REPLACE USING フローとして定義し、指定されたキーカラムに一致するターゲットテーブルのすべての行を置き換え、他の行はそのままにします。 REPLACE USING元が列ごとにキー化された部分的なスナップショットの連続であれば使います。 SEQUENCE BY 更新の順序は、キーの最高順位が勝つようにします。たとえ更新が順不同であってもです。

少なくとも1つのキーカラムと、正確に1つの SEQUENCE BY カラムを指定します。クエリはストリーミングクエリでなければならず、 BY NAME が必要です。 REPLACE USING ONCEやAUTO CDC ... INTOと組み合わせることはできない。

詳細については、「部分スナップショット置換」と「REPLACE USING フロー」を参照してください。
条件 WHERE 置き換え

フローを REPLACE WHERE フローとして定義し、ターゲットテーブルのターゲット部分集合を再計算し上書きします。各更新の際に、ターゲットテーブル内の condition に一致するすべての行が削除され、同じ述語範囲でソースクエリが再計算され、結果が挿入されます。 conditionと合わない列はそのまま残されます。ソースクエリに述語を追加する必要はありません。パイプラインエンジンはソースから読み取る際に自動的に適用します。

REPLACE WHERE バッチセマンティクスを使用しているため、ソースクエリがストリーミングクエリである必要はありません。 BY NAME は必須です。 REPLACE WHERE ONCE、REPLACE USING、AUTO CDC ... INTO と組み合わせることはできません。

詳細については、単独のストリーミングテーブルについては「 REPLACE WHERE flowsを用いたバッチ処理」および「REPLACE WHERE flows」を参照してください。
ある時

必要に応じて、バックフィルなどの 1 回限りのフローとしてフローを定義します。 ONCEを使用すると、次の 2 つの方法でフローが変更されます。
- ソース query または create_auto_cdc_flow_spec はストリーミングテーブルではありません。
- フローは既定で 1 回実行されます。パイプラインが完全な更新で更新された場合、 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-08

CREATE FLOW (パイプライン)

構文

パラメーター

例示

フィードバック

その他のリソース