CREATE STREAMING TABLE (パイプライン)

ストリーミングテーブルは、ストリーミングまたは増分データ処理をサポートするテーブルです。ストリーミングテーブルはパイプラインによってサポートされます。ストリーミングテーブルが更新されるたびに、ソーステーブルに追加されたデータがストリーミングテーブルに追加されます。ストリーミングテーブルは、手動で、またはスケジュールに従って更新できます。

更新を実行またはスケジュールする方法の詳細については、「パイプライン更新の実行」を参照してください。

構文

CREATE [OR REFRESH] [PRIVATE] STREAMING TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ {flow_clause | AS query} ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ column_constraint ] [, ...]
    [ , table_constraint ] [...] )

   column_properties
      { NOT NULL | GENERATED ALWAYS AS ( expr ) | GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( [ START WITH start | INCREMENT BY step ] [ ...] ) ] | DEFAULT default_expression | COMMENT column_comment | column_constraint | MASK clause } [ ... ]

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

flow_clause
  FLOW { { INSERT [ONCE] BY NAME query } |
  { AUTO CDC auto_cdc_flow_spec } |
  { AUTO CDC [ONCE] auto_cdc_from_snapshot_spec } |
  { REPLACE WHERE predicate BY NAME query } |
  { REPLACE USING ( column_name [, ...] ) SEQUENCE BY sequence_column BY NAME query } }

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 ) } ]

パラメーター

REFRESH

指定した場合は、テーブルを作成するか、既存のテーブルとその内容を更新します。
プライベート

プライベートストリーミングテーブルを作成します。
- これらはカタログに追加されず、定義したパイプライン内でのみアクセスできます。
- カタログ内の既存のオブジェクトと同じ名前を持つことができます。パイプライン内で、プライベートストリーミングテーブルとカタログ内のオブジェクトの名前が同じである場合、その名前への参照はプライベートストリーミングテーブルに解決されます。
- プライベートストリーミングテーブルは、1 回の更新中ではなく、パイプラインの有効期間全体でのみ保持されます。
プライベートストリーミングテーブルは、以前は TEMPORARY パラメーターを使用して作成されました。
table_name

新しく作成されたテーブルの名前。完全修飾のテーブル名は一意にする必要があります。
テーブル仕様

この省略可能な句で、列、その型、プロパティ、説明、および列制約の一覧を定義します。
- column_identifier
  
  列名は一意である必要があり、かつクエリの出力列にマップされている必要があります。
- カラムタイプ (column_type)
  
  列のデータ型を指定します。 Azure Databricks でサポートされているすべてのデータ型が、ストリーミングテーブルでサポートされているわけではありません。
- column_comment
  
  列を記述する任意のSTRING リテラル。このオプションは、column_type と共に指定する必要があります。列の種類が指定されていない場合、列コメントはスキップされます。
- 常に自動生成 ( expr )
  
  この句を指定すると、この列の値は、指定した expr によって決定されます。
  
  テーブルの DEFAULT COLLATION は UTF8_BINARYする必要があります。
  
  expr は、以下のものを除く、リテラル、テーブル内の列識別子、および決定論的な組み込みの SQL 関数または演算子で構成される場合があります。
  - 集計関数
  - 分析ウィンドウ関数
  - 順位付けウィンドウ関数
  - テーブル値ジェネレーター関数
  - UTF8_BINARY 以外の照合順序を持つ列
  また expr には、サブクエリを含めることはできません。
- GENERATED { ALWAYS |既定 } AS IDENTITY [ ( [ START WITH start ] [ INCREMENT BY step ] ) ] ]
  
  適用対象: Databricks SQL Databricks Runtime 10.4 LTS 以降
  
  ID 列を定義します。テーブルへの書き込み時に ID 列の値を指定しなかった場合は、統計的に増加する (または step が負の場合は減少する) 一意の値が自動的に割り当てられます。この句は、Delta テーブルでのみサポートされます。この句は、BIGINT データ型の列にのみ使用できます。
  
  自動的に割り当てられた値は、start から始まり、step ずつ増えます。割り当てられた値は一意ですが、連続している保証はありません。どちらのパラメーターも省略可能で、既定値は 1 です。 step に 0 は指定できません。
  
  自動的に割り当てられた値が ID 列の型の範囲を超える場合、クエリは失敗します。
  
  ALWAYS を使用する場合は、ID 列に独自の値を指定できません。
  
  次の操作はサポートされていません。
  - ID 列の PARTITIONED BY を行う
  - ID 列の UPDATE を行う
  注
  
  テーブルで ID 列を宣言すると、同時実行トランザクションが無効になります。 ID 列は、ターゲットテーブルへの同時書き込みが不要なユースケースでのみ使用してください。
- デフォルトデフォルト式
  
  適用対象: Databricks SQL Databricks Runtime 11.3 LTS以上
  
  列が指定されていない場合に DEFAULT、INSERT、UPDATE で使われる列の MERGE ... INSERT 値を定義します。
  
  既定値が指定されていない場合、Null 許容列には DEFAULT NULL が適用されます。
  
  default_expression は、リテラル、および組み込みの SQL 関数か演算子で構成することができます。ただし、次のものは除きます。
  - 集計関数
  - 分析ウィンドウ関数
  - 順位付けウィンドウ関数
  - テーブル値ジェネレーター関数
  また default_expression には、サブクエリを含めることはできません。
  
  DEFAULT は CSV、JSON、PARQUET、と ORC ソースでサポートされています。
- column_constraint
  
  ストリーミングテーブルの列に、情報主キー制約または情報外部キー制約を追加します。
- MASK 句
  
  列マスク関数を追加して、機密データを匿名化します。
  
  「行フィルターと列マスク」を参照してください。
- CONSTRAINT expectation_name EXPECT (expectation_expr) [ 違反時 { FAIL UPDATE | ROWを削除 } ]
  
  ストリーミングテーブルにデータ品質の期待値を追加します。これらのデータ品質の期待値は、時間の経過と同時に追跡し、ストリーミングテーブルのイベントログを介してアクセスできます。テーブルの作成時とテーブル更新時の両方で、FAIL UPDATE 期待値により処理が失敗します。 DROP ROW 期待値が満たされない場合、行全体が削除されます。パイプラインの期待を使用してデータ品質を管理する方法については、を参照してください。
  
  expectation_expr は、以下のものを除く、リテラル、テーブル内の列識別子、および決定論的な組み込みの SQL 関数または演算子で構成される場合があります。
  - 集計関数
    - 分析ウィンドウ関数
    - 順位付けウィンドウ関数
    - テーブル値ジェネレーター関数
  また expr には、サブクエリを含めることはできません。
テーブル制約

スキーマを指定するときに、主キーと外部キーを定義できます。制約は情報提供のみを目的としており、強制されるものではありません。 SQL 言語リファレンスの CONSTRAINT 句を参照してください。

注

テーブル制約を定義するには、パイプラインが Unity Catalog 対応のパイプラインである必要があります。
テーブル条項

必要に応じて、テーブルのパーティション分割、コメント、ユーザー定義プロパティを指定します。各サブ句は、1 回だけ指定できます。
- DELTA の使用
  
  データ形式を指定します。オプションは DELTA のみです。
  
  この句は省略可能で、既定値は DELTA です。
- でパーティション分割
  
  テーブル内のパーティション分割に使用する 1 つ以上の列のリスト (省略可能)。 CLUSTER BY と相互に排他的です。
  
  液体クラスタリングは、クラスタリング用の柔軟で最適化されたソリューションを提供します。パイプラインにCLUSTER BYするのではなく、PARTITIONED BYを使用することを検討してください。
- CLUSTER BY
  
  リキッドクラスタリングをテーブルに対して有効化し、クラスタリングキーとして使用する列を定義します。 CLUSTER BY AUTOで自動液体クラスタリングを使用し、Databricks はクエリのパフォーマンスを最適化するためにクラスタリングキーをインテリジェントに選択します。 PARTITIONED BY と相互に排他的です。
  
  表に液体クラスタリングを使用するを参照してください。
- 場所
  
  テーブルデータの保存場所 (省略可能)。設定されていない場合、システムは既定でパイプラインストレージの場所に設定されます。
- コメント
  
  テーブルについて説明するオプションの STRING リテラル。
- TBLPROPERTIES
  
  テーブルのテーブルプロパティのリスト (省略可能)。
- で ROW FILTER
行フィルター関数をテーブルに追加します。それ以降のそのテーブルに対するクエリでは、行のうち、この関数による評価の結果が TRUE であるものだけが返されます。これは、細粒度のアクセス制御に役立ちます。呼び出し元ユーザーの ID とグループメンバーシップをその関数で検査した結果として、特定の行をフィルター処理するかどうかを決定できるからです。

ROW FILTER の条項を参照してください。
- フロー
  
  必要に応じて、テーブルを作成するフローをインラインで定義します。フローは、テーブルの内容を更新するステートフルクエリです。 FLOWを指定しない場合は、代わりにAS queryを使用するか、CREATE FLOWを使用してフローを個別に定義できます。次のいずれかのフローの種類を指定できます。
  - INSERT 名前別
    
    列名でテーブルにデータを挿入します。 ONCE オプションを指定しない場合、クエリはストリーミングクエリである必要があります。ストリーミングセマンティクスを使用してソースから読み取る場合は、 STREAM キーワードを使用します。読み取りで既存のレコードの変更または削除が発生した場合は、エラーがスローされます。静的ソースまたは追加専用ソースから読み取るのが最も安全です。
    注
    
    FLOW INSERT BY NAME は、 AS queryの使用と同じです。次の 2 つのステートメントの動作は同じです。
```
CREATE OR REFRESH STREAMING TABLE raw_data
AS SELECT * FROM STREAM read_files('abfss://my_path');

CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');
```
  - ある時
    
    必要に応じて、バックフィルなどの 1 回限りのフローとしてフローを定義します。 ONCEが指定されると、クエリはストリーミングクエリではなく、フローは既定で 1 回実行されます。テーブルが完全更新で更新されると、 ONCE フローが再度実行され、データが再作成されます。 ONCE INSERT BY NAMEおよびAUTO CDC ... FROM SNAPSHOTフローに適用されます。
    
    WITH VERSION AUTO CDC ... FROM SNAPSHOTフローの場合、各更新はバージョンクエリが行を返さないまで順番に処理します。 ONCEと非ONCEの場合は、後のアップデートで何が起こるかが異なります。
    - ONCEがなければ、フローは毎回のアップデートで動作します。各更新は新しいバージョンを処理し、その後 IDLE に戻り、後期バージョンが現れるかテーブルが完全に更新されるまで続きます。
    - ONCEの場合:フローは最初の成功したスナップショットコミット後に完了し、その後の増分更新ではスキップされます。最初のバージョンクエリが行を返さず、スナップショットコミットも返さないため、フローは完了せずに適格のままになります。フルリフレッシュで再度実行します。
  - AUTO CDC
    
    Important
    
    Databricks Runtime 17.3 以降および PREVIEW Pipelines チャネルで使用できます。
    
    ソースからテーブルへの変更データキャプチャ (CDC) レコードを処理する AUTO CDC フローを定義します。ソースデータに CDC セマンティクスが含まれている場合は、 AUTO CDC を使用します。「AUTO CDC API: パイプラインを使用して変更データキャプチャを簡略化する」を参照してください。
  - AUTO CDC ... FROM SNAPSHOT
    
    変更フィードを読むのではなく、スナップショットを比較して変更を導き出すインライン AUTO CDC フローを定義します。ソースで変更データキャプチャが有効になっておらず、完全なスナップショットのみが利用可能の場合は、このフォームを使用してください。ソースは必須の FROM SNAPSHOT (snapshot_query) 節とオプションの WITH VERSION (version_query) 節で指定されています。この埋め込みフローは匿名で、ストリーミングテーブルが所有し、 INTO target 節はありません。スタンドアロン形式と同じ、 ONCE、 FROM SNAPSHOT (...)、 WITH VERSION (...)、 KEYS、 STORED AS、 TRACK HISTORY ON パラメータを受け入れています。詳細な説明は「CREATE FLOW (pipelines)」を参照してください。また、「 AUTO CDC from SNAPSHOT」の仕組みも参照してください。
    
    ソースは括弧内の FROM SNAPSHOT (query) フォームを使用しなければなりません。以前の単一識別子 FROM SNAPSHOT sourceを移行するには、 FROM SNAPSHOT (SELECT * FROM source)として書き換えてください。
  - REPLACE WHERE NAME クエリで指定する
    
    他のすべての行を変更せずに、REPLACE WHEREに一致する行のみを再計算して上書きするpredicate フローを定義します。結合と集計、到着遅延データ、スキーマの進化、バックフィルの増分バッチ処理には、 REPLACE WHERE を使用します。 BY NAME は必須です。 REPLACE WHERE フローを使用したバッチ処理を参照してください。
  - 使用を置き換えてください( column_name [、...])名前によるsequence_column順クエリ
    
    Important
    
    この機能はベータ版です。 Databricks Runtime 18.2以上が必要です。
    
    指定されたキーカラムに一致するすべての行を置き換え、他のすべての行はそのままにする REPLACE USING フローを定義します。 REPLACE USING元が列ごとにキー化された部分的なスナップショットの連続であれば使います。 SEQUENCE BY 更新の順序は、キーの最高順位が勝つようにします。たとえ更新が順不同であってもです。ソースはストリーミングソースである必要があります。 BY NAME は必須です。部分スナップショット置換はREPLACE USING flowsで参照してください。
AS クエリ

この句により、query からデータがテーブルに入力されます。このクエリはストリーミング クエリにする必要があります。 STREAM キーワードを使用して、ストリーミングセマンティクスを使用してソースから読み取ります。読み取りで既存のレコードの変更または削除が発生した場合は、エラーがスローされます。静的ソースまたは追加専用ソースから読み取るのが最も安全です。変更コミットがあるデータを取り込むには、 skipChangeCommits 読み取りオプションを追加してエラーを処理できます。

query と table_specification を一緒に指定するとき、table_specification に指定されているテーブルスキーマに、query から返される列をすべて含める必要があります。含まれていない場合、エラーが出ます。 table_specification で指定されているが、query から返されない列はクエリ時に null 値を返します。

ストリーミングデータの詳細については、「パイプラインを使用してデータを変換する」を参照してください。
- 読み取りオプション
  
  クエリで読み取りオプションを指定して、ソースからデータを読み取る方法を構成できます。たとえば、ソースデータ内の変更コミットをスキップする skipChangeCommits を指定できます。読み取りオプションは、クエリの WITH 句でマップとして指定されます。例えば次が挙げられます。
```
SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS=TRUE, STARTINGVERSION=X)
```
  =TRUEは省略可能であるため、次のようなブール値オプションを指定することもできます。
```
SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS)
```
  注
  
  読み取りオプションは、Databricks Runtime 17.3 以降でのみサポートされています。
  
  以下の読み取りオプションは Delta でサポートされています。各オプションの詳細については、 Delta Lake テーブルストリーミングの読み取りと書き込みを参照してください。
  - maxFilesPerTrigger
  - maxBytesPerTrigger
  - startingVersion
  - startingTimestamp
  - readChangeFeed
  - withEventTimeOrder
  - skipChangeCommits

必要なアクセス許可

パイプラインの実行時のユーザーには、次のアクセス許可が必要です。

ストリーミングテーブルによって参照されるベーステーブルに対する SELECT 権限。
親カタログに対する USE CATALOG 特権と、親スキーマに対する USE SCHEMA 特権。
ストリーミングテーブルのスキーマに対する CREATE MATERIALIZED VIEW 権限。

ストリーミングテーブルが定義されているパイプラインをユーザーが更新できるようにするには、次が必要です。

親カタログに対する USE CATALOG 特権と、親スキーマに対する USE SCHEMA 特権。
ストリーミングテーブルの所有権またはストリーミングテーブルに対する REFRESH 権限。
ストリーミングテーブルの所有者は、ストリーミングテーブルによって参照されるベーステーブルに対する SELECT 権限を持っている必要があります。

ユーザーが結果のストリーミングテーブルに対してクエリを実行できるようにするには、次が必要です。

親カタログに対する USE CATALOG 特権と、親スキーマに対する USE SCHEMA 特権。
ストリーミングテーブルに対する SELECT 権限。

制限事項

テーブル所有者だけがストリーミングテーブルを更新して最新のデータを取得できます。
ALTER TABLE コマンドはストリーミングテーブルでは許可されません。テーブルの定義とプロパティは、CREATE OR REFRESH または ALTER STREAMING TABLE ステートメントを使用して変更する必要があります。
INSERT INTO や MERGE などの DML コマンドを利用してテーブルスキーマを導き出すことはできません。
次のコマンドは、ストリーミングテーブルではサポートされていません。
- CREATE TABLE ... CLONE <streaming_table>
- COPY INTO
- ANALYZE TABLE
- RESTORE
- TRUNCATE
- GENERATE MANIFEST
- [CREATE OR] REPLACE TABLE
テーブルの名前変更や所有者の変更はサポートされていません。

例示

-- Define a streaming table from a volume of files:
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- Define a streaming table from a streaming source table:
CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(customers_bronze)

-- Use automatic liquid clustering to let Databricks choose the clustering columns:
CREATE OR REFRESH STREAMING TABLE customers_bronze_auto
CLUSTER BY AUTO
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- Define a table with a row filter and column mask:
CREATE OR REFRESH STREAMING TABLE customers_silver (
  id int COMMENT 'This is the customer ID',
  name string,
  region string,
  ssn string MASK catalog.schema.ssn_mask_fn COMMENT 'SSN masked for privacy'
)
WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
AS SELECT * FROM STREAM(customers_bronze)

-- Define a streaming table with an identity column:
CREATE OR REFRESH STREAMING TABLE customers_with_id (
  customer_id BIGINT GENERATED ALWAYS AS IDENTITY,
  name string,
  region string
)
AS SELECT name, region FROM STREAM(customers_bronze)

-- Define a streaming table that you can add flows into:
CREATE OR REFRESH STREAMING TABLE orders;

-- Define a streaming table with an inline append flow:
CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');

-- Define a streaming table with an inline AUTO CDC flow:
CREATE OR REFRESH STREAMING TABLE target
FLOW AUTO CDC
FROM stream(cdc_data.users)
KEYS (userId)
SEQUENCE BY sequenceNum
STORED AS SCD TYPE 1;

-- Define a streaming table with an inline REPLACE USING flow that keeps the latest
-- row for each payment_id:
CREATE OR REFRESH STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

-- Define a streaming table with an inline AUTO CDC FROM SNAPSHOT flow that processes
-- each new snapshot version on every update, using WITH VERSION to pick the next one:
CREATE STREAMING TABLE users
FLOW AUTO CDC
FROM SNAPSHOT (
  SELECT * FROM read_files('/Volumes/catalog/schema/landing/users/', 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/users/')
  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 (user_id)
STORED AS SCD TYPE 1;

-- Define a streaming table with an inline one-shot AUTO CDC FROM SNAPSHOT backfill:
CREATE STREAMING TABLE customers
FLOW AUTO CDC ONCE
FROM SNAPSHOT (SELECT * FROM catalog.schema.customers_snapshot)
KEYS (customer_id)
STORED AS SCD TYPE 1;

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-05