CREATE STREAMING TABLE (パイプライン)

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

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

構文

CREATE [OR REFRESH] [PRIVATE] STREAMING TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ AS query ]

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

   column_properties
      { NOT NULL | 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 } } [ ... ]

パラメーター

REFRESH

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

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

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

この省略可能な句で、列、その型、プロパティ、説明、および列制約の一覧を定義します。
- column_identifier
  
  列名は一意である必要があり、かつクエリの出力列にマップされている必要があります。
- カラムタイプ (column_type)
  
  列のデータ型を指定します。 Azure Databricks でサポートされているすべてのデータ型が、ストリーミングテーブルでサポートされているわけではありません。
- column_comment
  
  列を記述する任意のSTRING リテラル。このオプションは、column_type と共に指定する必要があります。列の種類が指定されていない場合、列コメントはスキップされます。
- column_constraint
  
  テーブルに流入するデータを検証する制約を追加します。パイプラインの期待を使用してデータ品質を管理する方法については、を参照してください。
- MASK 句
  
  Important
  
  この機能はパブリックプレビュー段階です。
  
  列マスク関数を追加して、機密データを匿名化します。
  
  「行フィルターと列マスク」を参照してください。
テーブル制約

Important

この機能はパブリックプレビュー段階です。

スキーマを指定するときに、主キーと外部キーを定義できます。制約は情報提供のみを目的としており、強制されるものではありません。 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
Important

この機能はパブリックプレビュー段階です。

行フィルター関数をテーブルに追加します。それ以降のそのテーブルに対するクエリでは、行のうち、この関数による評価の結果が TRUE であるものだけが返されます。これは、細粒度のアクセス制御に役立ちます。呼び出し元ユーザーの ID とグループメンバーシップをその関数で検査した結果として、特定の行をフィルター処理するかどうかを決定できるからです。

ROW FILTER の条項を参照してください。
クエリ

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

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

ストリーミングデータの詳細については、「パイプラインを使用してデータを変換する」を参照してください。

必要なアクセス許可

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

ストリーミングテーブルによって参照されるベーステーブルに対する 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
テーブルの名前変更や所有者の変更はサポートされていません。
生成された列、ID 列、既定の列はサポートされていません。

例示

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

-- 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 that you can add flows into:
CREATE OR REFRESH STREAMING TABLE orders;

フィードバック

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

Last updated on 2025-11-11