CREATE TABLE ...FLOW (パイプライン)

Important

この機能は ベータ版です。

CREATE TABLE ... FLOW ステートメントを使用して、1 つ以上のフローによって記述されたマネージド テーブルをパイプラインに作成します。

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

複数のソースを 1 つのマネージド テーブルにファンするには、 CREATE FLOW (パイプライン) を使用してそれをターゲットとする複数のフローを宣言します。

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    作成するマネージド テーブルの名前。 名前が修飾されていない場合、テーブルはパイプラインのターゲット スキーマに作成されます。 この名前は、ストリーミング テーブルにまだ属していない必要があります。

  • テーブル仕様

    必要に応じて、列、その型、プロパティ、および説明を定義します。 省略した場合、スキーマはフロー クエリから推論されます。

  • CONSTRAINT expectation_name EXPECT (expectation_expr) [ 違反時 { FAIL UPDATE | ROWを削除 } ]

    管理されたテーブルにデータ品質の期待値を追加します。 これらのデータ品質の期待値は時間経過とともに追跡でき、パイプラインの イベントログを通じてアクセス可能です。 テーブルの作成時とテーブル更新時の両方で、FAIL UPDATE 期待値により処理が失敗します。 DROP ROW 期待値が満たされない場合、行全体が削除されます。 「パイプラインの期待値を使用してデータ品質を管理する」を参照してください。

    expectation_expr リテラル、テーブル内の列識別子、決定的で組み込みのSQL関数や演算子で構成されることがありますが、以下を除きます:

    また expectation_expr には、サブクエリを含めることはできません。

  • (col [, ...])で分割

    必要に応じて、列のサブセットによってテーブルをパーティション分割します。

  • CLUSTER BY 条項

    必要に応じて、テーブルで 液体クラスタリング を有効にします。 PARTITIONED BYとCLUSTER BYを組み合わせることはできません。

  • 場所のパス

    テーブル データの保存場所 (省略可能)。

  • コメント table_comment

    テーブルを記述する STRING リテラル。

  • TBLPROPERTIES 句

    必要に応じて、1 つ以上のユーザー定義テーブル プロパティを設定します。

  • WITH ROW FILTER 句

    行フィルター関数をテーブルに追加します。 そのテーブルの今後のクエリは、関数が TRUE評価する行のサブセットを受け取ります。

  • FLOW INSERT [ONCE] BY NAME クエリ

    テーブルに query の結果を挿入し、結果列をテーブル列に 名前で一致させる追加フローを定義します。 query は、バッチ ソースまたはストリーミング ソースを参照できます。 ONCE は、すべての更新ではなく、フローを 1 回 (バックフィルなど) 実行します。 各名前付きフローは、パイプラインの更新ごとに 1 回だけ入力を処理します。ストリーミング テーブルの FLOW INSERT BY NAME と同じです。

  • 輸入検問checkpoint_path

    既存の構造化ストリーミングチェックポイントへのパスを STRING リジラルで、フローにインポートするため、移行したストリームはソースを再処理する代わりに、最後にコミットされたオフセットから再開します。 チェックポイントのインポートは ベータ版です。 詳細は 「構造化ストリーミングチェックポイントの移行」を参照してください。

構造化ストリーミングチェックポイントの移行

Important

チェックポイントのインポートは ベータ版です。

IMPORT CHECKPOINTを使って、既存の構造化ストリーミングワークロードをソースを再処理せずにパイプラインに移行できます。 構造化ストリーミングクエリの使ったものcheckpointLocationにcheckpoint_path設定してください。クラウドストレージ、Unity Catalogボリューム、DBFSパスなどです。 最初のパイプライン更新では、そのチェックポイントをパイプラインの管理ストレージにクローンします。 その後、フローは最後にコミットされたオフセットから再開され、その状態(アグリゲーション、重複除去キー、ウォーターマークなど)がそのままです。 その後のパイプライン更新では、フローのクローンチェックポイントを使用します。元のチェックポイントは変更されません。

Warning

パイプラインを実行する前に、元のStructured Streamingクエリを停止してください。 元のStructured Streamingクエリはインポート後に再利用できますが、チェックポイントの状態を管理し、パイプラインとクエリが同時に同じテーブルに書き込みないようにする必要があります。重複したデータが発生する可能性があります。

構造化ストリーミングクエリをパイプラインフローとして再作成し、クエリが書き込んだテーブルに書き込み、チェックポイントをインポートします。

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

チェックポイントは最初のパイプライン更新時にのみインポートされます。 後のアップデートでは IMPORT CHECKPOINTが無視されます。 完全なリフレッシュはチェックポイントを再インポートしません。 新しい空のチェックポイントから出発し、ソースを再処理します。

制限事項

  • マネージド テーブルでは、CDC 変更フローはサポートされていません。 AUTO CDC INTO(SQL) またはマネージド テーブルに対するapply_changes / apply_changes_from_snapshot (Python) は、MANAGED_TABLE_DOES_NOT_SUPPORT_CDCで失敗します。 CDCのターゲットには、CREATE FLOW(パイプライン)で定義されたフローを持つCREATE STREAMING TABLE(パイプライン)を使用します。
  • マネージド テーブルでは、 FLOW ... REPLACE WHEREはサポートされていません。 FLOW INSERT BY NAME のみがサポートされています。 テーブルのターゲットとなるサブセットを再計算・上書きするには、ストリーミングテーブルを作成し、CREATE FLOW(パイプライン)でREPLACE WHEREフローを定義します。 REPLACE WHERE フローを使用したバッチ処理を参照してください。
  • マネージド テーブルは、Unity カタログを使用するパイプラインでのみサポートされます。 Hive メタストアはサポートされていません。
  • マネージド テーブルの既存のストリーミング テーブルの名前を再利用することはできません。 ストリーミング テーブルを最初に削除するか、ステートメントが CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLEで失敗します。
  • IMPORT CHECKPOINT FLOW INSERT BY NAMEのみ対応しています。

例示

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

その他のリソース

  • 最終更新: