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 [ IMPORT CHECKPOINT checkpoint_path ] 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 と同様に、パイプラインの更新ごとにその入力を正確に一度だけ処理します。

• IMPORT CHECKPOINT checkpoint_path

  フローにインポートする既存のStructured Streamingチェックポイントへのパスを含む STRING リテラル。移行されたストリームは、ソースを再処理するのではなく、最後にコミットされたオフセットから再開されます。チェックポイントのインポートはベータ版です。Migrate a Structured Streaming checkpointを参照してください。

Structured Streaming チェックポイントの移行

備考

ベータ版

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

ソースを再処理せずに既存の Structured Streaming ワークロードをパイプラインに移行するには、IMPORT CHECKPOINTを使用します。checkpoint_pathを Structured Streaming クエリーが使用したcheckpointLocationに設定します。これは、クラウドストレージ、Unity Catalog ボリューム、または DBFS パスにすることができます。最初のパイプライン更新時に、フローはそのチェックポイントをパイプラインの管理対象ストレージに複製します。その後、フローは、その状態（集計、重複排除キー、ウォーターマークなど）を保持したまま、最後にコミットされたオフセットから再開します。後続のパイプライン更新では、フローの複製されたチェックポイントが使用されます。元のチェックポイントは変更されません。

警告

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

Structured Streaming クエリーを、クエリーが書き込んだテーブルに書き込み続けるパイプライン フローとして再作成し、チェックポイントをインポートします：

SQL

-- 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 Catalog を使用したパイプラインでのみサポートされています。Hive metastore はサポートされていません。
• 既存のストリーミングテーブルの名前をマネージドテーブルに再利用することはできません。まずストリーミングテーブルを削除してください。そうしないと、ステートメントはCANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLEで失敗します。
• IMPORT CHECKPOINT は FLOW INSERT BY NAME でのみサポートされています。

例

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

その他のリソース

• CREATE STREAMING TABLE (パイプライン)
• CREATE FLOW (パイプライン)
• CREATE MATERIALIZED VIEW (パイプライン)
• パイプラインSQL言語リファレンス
• LakeFlow Pipelines Python 言語リファレンス