スタンドアロンストリーミングテーブルの使用

スタンドアロンの ストリーミングテーブル は、Lakeflowパイプラインの外部で定義された、ストリーミングまたは増分データ処理の追加サポートを備えた Unity Catalog に登録されたテーブルです。パイプラインは、ストリーミングテーブルごとに自動的に作成されます。ストリーミングテーブルを使用して、Kafka およびクラウドオブジェクトストレージからの増分データロードを行うことができます。

Databricks SQL warehouse、またはサーバレス汎用コンピュートで実行されているノートブックから、スタンドアロンのストリーミングテーブルを作成および更新できます。2 つのコンピュートオプションの違いの詳細については、スタンドアロンパイプラインの要件を参照してください。

ノートブックから Python を使用してスタンドアロンストリーミングテーブルを作成および更新するには、スタンドアロンパイプラインでの Python の使用を参照してください。

注記

Delta Lakeテーブルをストリーミングソースおよびシンクとして使用する方法については、Delta Lakeテーブルのストリーミング読み取りと書き込みを参照してください。

要件

スタンドアロンストリーミングテーブルの作成、更新、およびクエリに関するコンピュートオプション、アクセス許可、およびその他の要件については、スタンドアロンパイプラインの要件を参照してください。

ストリーミングテーブルの作成

ストリーミングテーブルは、Databricks SQLのSQLクエリによって定義されます。ストリーミングテーブルを作成すると、ソーステーブルに現在存在するデータを使用してストリーミングテーブルが作成されます。その後、通常はスケジュールに従ってテーブルを更新し、ソーステーブルに追加されたデータをプルしてストリーミングテーブルに追加します。

ストリーミングテーブルを作成するときは、そのテーブルの所有者と見なされます。

既存のテーブルからストリーミングテーブルを作成するには、次の例のように CREATE STREAMING TABLE ステートメントを使用します。

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT product, price FROM STREAM raw_data;

この場合、ストリーミングテーブル sales は raw_data テーブルの特定の列から作成され、1時間ごとに更新されるスケジュールが設定されます。使用するクエリーは ストリーミング クエリーである必要があります。STREAM キーワードを使用してストリーミングセマンティクスを使用し、ソースから読み取ります。

更新に使用されるコンピュート

CREATE OR REFRESH STREAMING TABLE ステートメントを使用してストリーミングテーブルを作成すると、初期データの更新とデータ作成がすぐに開始されます。これらの操作では、Databricks SQL warehouse コンピュートは消費されません。代わりに、ストリーミングテーブルは、作成と更新の両方をServerless パイプラインに依存しています。専用のServerless パイプラインは、ストリーミングテーブルごとにシステムによって自動的に作成および管理されます。

Auto Loaderでのファイルの読み込み

ボリューム内のファイルからストリーミングテーブルを作成するには、 Auto Loaderを使用します。Auto Loader を使用して、クラウドオブジェクトストレージからのほとんどのデータ取り込みタスクに対応します。Auto Loader とパイプラインは、クラウドストレージに到着すると、増え続けるデータを増分的かつべき等に読み込むように設計されています。

Databricks SQLでAuto Loaderを使用するには、read_files 関数を使用します。次の例は、 Auto Loader を使用して大量の JSON ファイルをストリーミングテーブルに読み込む方法を示しています。

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT * FROM STREAM read_files(
    "/Volumes/my_catalog/my_schema/my_volume/path/to/data",
    format => "json"
  );

クラウドストレージからデータを読み込むために、Auto Loaderも使用できます。

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
  's3://mybucket/analysis/*/*/*.json',
    format => "json"
  );

Auto Loaderの詳細については、Auto Loaderとはを参照してください。Auto LoaderでのSQL の使用について、例を挙げて詳しく知りたい場合は、オブジェクトストレージからのデータの読み込みを参照してください。

他のソースからのストリーミングインジェスト

他のソース (Kafkaを含む) からの取り込みの例については、パイプラインでのデータのロードを参照してください。

自動CDCフローによるチェンジデータキャプチャ (CDC) の適用

FLOW AUTO CDC 句を使用して、ソースからのチェンジデータキャプチャ（CDC）レコードをストリーミングテーブルに処理します。以前は、 MERGE INTO ステートメントはDatabricksでCDCレコードを処理するために一般的に使用されていました。しかし、 MERGE INTO は順不同のレコードのために誤った結果を生成したり、レコードを再順序付けするために複雑なロジックが必要になったりします。「チェンジデータキャプチャとスナップショット」を参照してください。

AUTO CDC 順不同のレコードを自動的に処理することでCDCを簡素化します。レコードを識別するキー、順序を指定するためのシーケンス列、および結果をSCDタイプ1（直接更新）として保存するかSCDタイプ2（履歴追跡）として保存するかを指定します。

次の例では、SCDタイプ1を使用してCDC変更を適用するストリーミングテーブルを作成します。

SQL
CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

次の例では、SCDタイプ2を使用して変更の履歴を保持します：

SQL
CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2;

Auto CDCオプションと動作の詳細については、「AUTO CDC APIs : パイプラインによるチェンジデータキャプチャの簡素化」を参照してください。構文リファレンスの完全版については、CREATE STREAMING TABLEを参照してください。

REPLACE WHEREフローを使用して選択的バッチ置換を適用する

FLOW REPLACE WHERE句を使用して、テーブル履歴全体を再処理せずにストリーミングテーブルのターゲットサブセットを再計算して上書きします。REPLACE WHEREフローは、結合と集計、データの遅延到着、アップストリームの再処理、スキーマ進化、バックフィルの増分バッチ処理に最適です。

要件、述語オーバーライド、増分更新など、REPLACE WHERE フローの詳細については、スタンドアロンストリーミングテーブルの REPLACE WHERE フローを参照してください。

REPLACE USING フローによる部分的なスナップショット置換の適用

備考

ベータ版

REPLACE USING フローはベータ版です。

FLOW REPLACE USING 句を使用して、ストリーミングテーブルを部分スナップショットのストリームと同期状態に保ちます。各更新で、REPLACE USINGフローは、指定されたキー列に一致するすべての行を置換し、他のすべての行はそのまま残します。SEQUENCE BY 列は更新を順序付けるため、更新が順不同で到着した場合でも、常にキーの最大シーケンスが優先されます。例:

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

BY NAME が必要です。列を位置ではなく名前で一致させます。

REPLACE USINGの動作は、スタンドアロンのストリーミングテーブルの場合も、LakeFlow Pipelinesの場合と同じです。その仕組み、シーケンス、期待値、制限、および例については、REPLACE USING フローによる部分的なスナップショットの置き換えを参照してください。スタンドアロンのストリーミングテーブルには、次の違いが適用されます。

SQLでフローを定義します。 CREATE OR REFRESH STREAMING TABLEのインラインSQLFLOW REPLACE USING句を使用して、REPLACE USINGフローを作成します。スタンドアロンのCREATE FLOW文はLakeflowパイプラインのコンストラクトであり、スタンドアロンのストリーミングテーブルには使用されません。
コンピュートはシステムによって管理されます。 スタンドアロンのストリーミングテーブルは、システム管理のServerlessパイプラインで実行され、Databricks Runtime 18.2以降が必要です。クラシックコンピュートとServerlessコンピュートのどちらかを選択する必要はありません。

新しいデータのみを取り込む

defaultでは、 read_files 関数はテーブル作成中にソースフォルダー内のすべての既存データを読み込み、その後、更新のたびに新しく到着するレコードを処理します。

テーブル作成時にソースフォルダにすでに存在するデータの取り込みを回避するには、includeExistingFilesオプションをfalseに設定します。これは、テーブル作成後にフォルダに到着したデータのみが処理されることを意味します。例えば：

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    '/path/to/files',
    includeExistingFiles => false
  );

ランタイムバージョン

ストリーミングテーブルは常に最新の Databricks SQL ランタイムバージョンで実行されます。以前に preview または current ランタイムチャンネルを選択するために使用されていた pipelines.channel テーブルプロパティは、サポートされなくなり、効果はありません。既存の定義にこのプロパティが含まれている場合、それは無視され、削除する必要はありません。

機密データを非表示にする

ストリーミングテーブルを使用すると、テーブルにアクセスするユーザーから機密データを隠すことができます。1つのアプローチは、機密性の高い列や行を完全に除外するようにクエリーを定義することです。代わりに、クエリーを実行するユーザーの権限に基づいて列マスクや行フィルターを適用できます。たとえば、HumanResourcesDeptグループに属していないユーザーに対してtax_id列を非表示にする場合などが考えられます。これを行うには、ストリーミングテーブルの作成中に ROW FILTER および MASK 構文を使用します。情報については、「行フィルターと列マスク」を参照してください。

ストリーミングテーブルの更新

ストリーミングテーブルは、Serverless パイプラインを自動的に作成して使用し、更新操作を処理します。更新はパイプラインによって管理され、更新はストリーミングテーブルの作成に使用された Databricks SQL warehouse によって監視されます。ストリーミングテーブルは、スケジュールに従って実行されるパイプラインを使用して更新できます。

スケジュールされた更新がある場合でも、いつでも手動で更新を呼び出すことができます。更新は、ストリーミングテーブルと一緒に自動的に作成されたのと同じパイプラインによって処理されます。

ストリーミングテーブルを更新するには：

SQL
REFRESH STREAMING TABLE sales;

DESCRIBE TABLE EXTENDEDを使用して、最新の更新のステータスを確認できます。

注記

タイムトラベルクエリーを使用する前に、ストリーミングテーブルを更新する必要がある場合があります。

更新のスケジュール方法については、更新のスケジュールを参照してください。定期的な更新には、更新通知を設定でき、更新のパフォーマンスモードを設定できます。

更新の仕組み

ストリーミングテーブルの更新では、最後の更新後に到着した新しい行のみが評価され、新しいデータのみが追加されます。

各更新では、ストリーミングテーブルの現在の定義を使用して、この新しいデータが処理されます。ストリーミングテーブルの定義を変更しても、既存のデータが自動的に再計算されるわけではありません。変更が既存のデータと互換性がない場合 (データ型の変更など)、次の更新はエラーで失敗します。

次の例では、ストリーミングテーブル定義への変更が更新動作にどのように影響するかを説明します。

フィルターを削除しても、以前にフィルターされた行は再処理されません。
列プロジェクションの変更は、既存データの処理方法には影響しません。
静的スナップショットとの結合は、最初の処理時のスナップショットの状態を使用します。更新されたスナップショットと一致するはずの遅れて到着したデータは無視されます。これにより、ディメンションが遅れた場合にファクトが破棄される可能性があります。
既存の列の CAST を変更するとエラーになります。

既存のストリーミングテーブルではサポートできない方法でデータが変更された場合は、完全な更新を実行できます。

ストリーミングテーブルのフルリフレッシュ

フル更新では、ソースで使用可能なすべてのデータが最新の定義で再処理されます。Kafkaのようにデータの履歴全体を保持していない、または保持期間が短いソースに対してフル更新を行うことは推奨されません。フル更新により既存のデータが切り捨てられるためです。ソースでデータが使用できなくなった場合、古いデータを回復できない可能性があります。

例えば：

SQL
REFRESH STREAMING TABLE sales FULL;

更新のスケジュールと監視

ストリーミングテーブルはスケジュールに従って、またはアップストリームデータが変更されたときに自動的に更新でき、更新のタイムアウト、通知、およびパフォーマンスモードを構成できます。更新のスケジュールを参照してください。

ストリーミングテーブルへのアクセスを制御する

ストリーミングテーブルは、豊富なアクセス制御をサポートして、潜在的にプライベートなデータを公開しないようにしながら、データ共有をサポートします。ストリーミングテーブルの所有者または MANAGE 権限を持つユーザーは、他のユーザーに SELECT 権限を付与できます。ストリーミングテーブルへの SELECT アクセス権を持つユーザーは、ストリーミングテーブルによって参照されるテーブルへの SELECT アクセス権を必要としません。このアクセス制御により、基になるデータへのアクセスを制御しながら、データ共有できます。

ストリーミングテーブルの所有者を変更することもできます。

ストリーミングテーブルに特権を付与する

ストリーミングテーブルへのアクセス権を付与するには、GRANTステートメントを使用します。

SQL
GRANT <privilege_type> ON <st_name> TO <principal>;

privilege_typeは次のとおりです：

SELECT - ユーザーはストリーミングテーブルを SELECT できます。
REFRESH - ユーザーはストリーミングテーブルを REFRESH できます。更新は所有者の権限を使用して実行されます。

次の例では、ストリーミングテーブルを作成し、ユーザーに選択権限と更新権限を付与します。

SQL
CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM STREAM source_table;

-- Grant read-only access:
GRANT SELECT ON st_name TO read_only_user;

-- Grant read and refresh access:
GRANT SELECT ON st_name TO refresh_user;
GRANT REFRESH ON st_name TO refresh_user;

Unity Catalogでセキュリティ保護可能なオブジェクトの権限を付与する方法の情報については、「Unity Catalog権限のリファレンス」を参照してください。

ストリーミングテーブルの権限の剥奪

ストリーミングテーブルからアクセス権を取り消すには、 REVOKEステートメントを使用します：

SQL
REVOKE privilege_type ON <st_name> FROM principal;

ソーステーブルに対する SELECT 特権がストリーミングテーブルの所有者やストリーミングテーブルに対して MANAGE または SELECT 特権を付与されている他のユーザーから取り消された場合、またはソーステーブルが削除された場合でも、ストリーミングテーブルの所有者やアクセス権を付与されたユーザーは引き続きストリーミングテーブルをクエリーできます。ただし、次の動作が発生します。

ストリーミングテーブルの所有者、またはストリーミングテーブルへのアクセス権を失った他のユーザーは、そのストリーミングテーブルをREFRESHできなくなり、ストリーミングテーブルは時間の経過とともに古くなります。
スケジュールによって自動化されている場合、次にスケジュールされたREFRESHは失敗するか、ランされません。

以下の例は、 read_only_user から SELECT 権限を取り消します：

SQL
REVOKE SELECT ON st_name FROM read_only_user;

ストリーミングテーブルの所有者を変更する

スタンドアロンストリーミングテーブルに対する MANAGE 権限を持つユーザーは、カタログエクスプローラを使用して新しい所有者を設定できます。新しい所有者は、自分自身または Service Principal ユーザー ロールを持つService Principalにすることができます。

Databricks ワークスペースから、[カタログ]をクリックして、カタログエクスプローラーを開きます。
更新するストリーミングテーブルを選択します。
右側のサイドバーの このストリーミングテーブルについて の下で、 所有者 を見つけて、編集をクリックします。

注記

パイプライン設定で ランユーザー を変更して所有者を更新するように求めるメッセージが表示された場合、ストリーミングテーブルはスタンドアロンテーブルとしてではなく、LakeFlow Pipelinesで定義されます。このメッセージには、パイプライン設定へのリンクが含まれており、そこで ランユーザー を変更できます。

ストリーミングテーブルの新しい所有者を選択します。

所有者は、所有するストリーミングテーブルに対する MANAGE および SELECT 権限を自動的に持っています。Service Principalを所有するストリーミングテーブルの所有者として設定し、ストリーミングテーブルに対する SELECT または MANAGE 権限を明示的に持っていない場合、この変更を行うと、ストリーミングテーブルへのすべてのアクセス権が失われることになります。この場合、これらの権限を明示的に付与するように求められます。

Grant MANAGE と Grant SELECT の両方の特権を選択して、 Save でそれらを提供します。
保存をクリックして、所有者を変更します。

ストリーミングテーブルの所有者が更新されました。今後の更新はすべて、新しい所有者の ID を使用して実行されます。

所有者がソーステーブルに対する権限を失った場合

所有者を変更し、新しい所有者がソーステーブルにアクセスできない場合（または基礎となるソーステーブルで SELECT 権限が取り消された場合）、ユーザーは引き続きストリーミングテーブルをクエリーできます。ただし：

ストリーミングテーブルをREFRESHできません。
ストリーミングテーブルの次回の予定された更新が失敗します。

ソースデータへのアクセスが失われると更新は防止されますが、既存のストリーミングテーブルが読み取られなくなるわけではありません。

ストリーミングテーブルからレコードを完全に削除する

備考

プレビュー

ストリーミングテーブルでの REORG ステートメントのサポートは、パブリックプレビュー段階です。

注記

ストリーミングテーブルで REORG ステートメントを使用するには、Databricks Runtime 15.4 以降が必要です。
任意のストリーミングテーブルで REORG ステートメントを使用できますが、削除ベクトルが有効なストリーミングテーブルからレコードを削除する場合にのみ必要です。削除ベクトルが有効になっていないストリーミングテーブルで使用した場合、このコマンドは効果がありません。

GDPRコンプライアンスなどのために、削除ベクトルが有効になっているストリーミングテーブルの基盤ストレージからレコードを物理的に削除するには、ストリーミングテーブルのデータに対してVACUUM操作が実行されるよう、追加のステップを実施する必要があります。

基礎となるストレージからレコードを物理的に削除するには：

ストリーミングテーブルからレコードを更新または削除します。
APPLY (PURGE)パラメーターを指定して、ストリーミングテーブルに対してREORGステートメントを実行します。たとえばREORG TABLE <streaming-table-name> APPLY (PURGE); 。
ストリーミングテーブルのデータ保持期間が経過するのを待ちます。defaultのデータ保持期間は7日間ですが、 delta.deletedFileRetentionDuration テーブルプロパティで構成できます。タイムトラベルクエリのデータ保持を構成するを参照してください。
REFRESH ストリーミングテーブル。ストリーミングテーブルの更新を参照してください。REFRESH操作から 24 時間以内に、レコードが完全に削除されるようにするために必要なVACUUM操作を含む、パイプラインメンテナンスタスクが自動的に実行されます。

クエリー履歴を使用して実行を監視する

クエリ履歴ページを使用して、クエリーの詳細とクエリープロファイルにアクセスできるため、ストリーミングテーブルの更新を実行するために使用されるパイプラインのパフォーマンスの低いクエリーやボトルネックを特定できます。クエリー履歴とクエリープロファイルで使用できる情報の種類の概要については、クエリー履歴とクエリープロファイルを参照してください。

備考

プレビュー

この機能はパブリックプレビュー段階です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

ストリーミングテーブルに関するすべてのステートメントは、クエリー履歴に表示されます。 ステートメント ドロップダウン・フィルターを使用して、任意のコマンドを選択し、関連するクエリーを調べることができます。すべての CREATE ステートメントの後には、パイプライン上で非同期に実行される REFRESH ステートメントが続きます。REFRESH ステートメントには、通常、パフォーマンスの最適化に関する知見を提供する詳細なクエリープランが含まれています。

クエリー履歴 UI で REFRESH ステートメントにアクセスするには、次のステップを実行します。

左のサイドバーにあるをクリックして、 クエリー履歴 UI を開きます。
ステートメント ドロップダウン・フィルターから REFRESH チェック・ボックスを選択します。
クエリー文の名前をクリックして、クエリーの期間や集計メトリクスなどの概要詳細を表示します。
[ クエリープロファイルの表示 ] をクリックして、クエリープロファイルを開きます。クエリープロファイルのナビゲーションの詳細については、「クエリープロファイル」を参照してください。
オプションで、 クエリーソース セクションのリンクを使用して、関連するクエリーまたはパイプラインを開くことができます。

SQLエディターのLink、またはSQL Warehouseにアタッチされたノートブックからクエリーの詳細にアクセスすることもできます。

外部クライアントからストリーミングテーブルにアクセスする

オープン APIs をサポートしていない外部 Delta Lake クライアントまたは Iceberg クライアントからストリーミングテーブルにアクセスするには、互換Modeを使用できます。互換モードは、任意の Delta Lake クライアントまたは Iceberg クライアントからアクセスできるストリーミングテーブルの読み取り専用バージョンを作成します。

要件​

ストリーミングテーブルの作成​

更新に使用されるコンピュート​

Auto Loaderでのファイルの読み込み​

他のソースからのストリーミング インジェスト​

自動CDCフローによるチェンジデータキャプチャ (CDC) の適用​

REPLACE WHEREフローを使用して選択的バッチ置換を適用する​

REPLACE USING フローによる部分的なスナップショット置換の適用​

新しいデータのみを取り込む​

ランタイムバージョン​

機密データを非表示にする​

ストリーミングテーブルの更新​

更新の仕組み​

ストリーミングテーブルのフルリフレッシュ​

更新のスケジュールと監視​

ストリーミングテーブルへのアクセスを制御する​

ストリーミングテーブルに特権を付与する​

ストリーミングテーブルの権限の剥奪​

ストリーミングテーブルの所有者を変更する​

所有者がソーステーブルに対する権限を失った場合​

ストリーミングテーブルからレコードを完全に削除する​

クエリー履歴を使用して実行を監視する​

外部クライアントからストリーミングテーブルにアクセスする​

その他のリソース​

要件