メインコンテンツまでスキップ
に最終更新

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

備考

ベータ版

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

CREATE TABLE ... FLOW ステートメントを使用して、1つまたは複数のフローによって書き込まれる**パイプライン**内に**マネージドテーブル**を作成します。

構文​

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 INSERT [ONCE] BY NAME query

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

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

パラメーター​

  • table_name

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

  • table_specification

    オプションで、列、その型、プロパティ、および説明を定義します。省略した場合、スキーマはフロークエリーから推論されます。

  • CONSTRAINT expectation_name EXPECT (expectation_expr) [ ON VIOLATION { FAIL UPDATE | DROP ROW } ]

    マネージドテーブルにデータ品質のエクスペクテーションを追加します。これらのデータ品質のエクスペクテーションは、時間の経過と共に追跡し、パイプラインの event Logsを通じてアクセスできます。FAIL UPDATEエクスペクテーションがあると、テーブルの作成時とテーブルの更新時の両方で処理が失敗します。DROP ROWエクスペクテーションでは、エクスペクテーションが満たされない場合、行全体が削除されます。パイプラインのエクスペクテーションを使用してデータ品質を管理するを参照してください。

    expectation_expr リテラル、テーブル内の列識別子、および決定論的な組み込み SQL 関数または演算子で構成できます（以下を除く）。

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

  • PARTITIONED BY（col [, ...]）

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

  • クラスター BY 句

    テーブルでリキッドクラスタリングをオプションで有効にします。PARTITIONED BYとCLUSTER BYを組み合わせることはできません。

  • 格納場所パス

    テーブルデータのオプションの保存場所です。

  • COMMENT テーブルコメント

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

  • TBLPROPERTIES句

    オプションで1つ以上のユーザー定義テーブルプロパティを設定します。

  • WITH 行フィルター句

    テーブルに行フィルター関数を追加します。そのテーブルに対する今後のクエリーでは、関数がTRUEと評価される行のサブセットが返されます。

  • フロー挿入 [一度] 名前でクエリー

    query の結果をテーブルに挿入するアペンドフローを定義し、結果列をテーブル列に 名前で 一致させます。query はバッチまたはストリーミングのソースを参照できます。ONCE は、すべての更新時ではなく、フローを単一回（たとえば、バックフィルのため）実行します。各名前付きフローは、ストリーミングテーブルの FLOW INSERT BY NAME と同様に、パイプラインの更新ごとにその入力を正確に一度だけ処理します。

制限事項​

  • マネージドテーブルはCDC変更フローをサポートしていません。マネージドテーブルに対する AUTO CDC INTO (SQL) または apply_changes / apply_changes_from_snapshot (Python) は MANAGED_TABLE_DOES_NOT_SUPPORT_CDC で失敗します。CDCターゲットには、CREATE STREAMING TABLE（パイプライン）を使用します。
  • マネージドテーブルはFLOW ... REPLACE WHEREをサポートしていません。FLOW INSERT BY NAMEのみサポートされています。
  • マネージドテーブルは、Unity Catalog を使用したパイプラインでのみサポートされています。Hive metastore はサポートされていません。
  • 既存のストリーミングテーブルの名前をマネージドテーブルに再利用することはできません。まずストリーミングテーブルを削除してください。そうしないと、ステートメントはCANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLEで失敗します。

例​

SQL
-- 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');

その他のリソース​

このページの見出し