Databricks SQLのストリーミングテーブルを使用してデータを読み込む

Databricks では、ストリーミングテーブルを使用して Databricks SQL を使用してデータを取り込むことをお勧めします。 ストリーミングテーブルは、Unity Catalog に登録されたテーブルで、ストリーミングまたは増分データ処理の追加サポートがあります。Delta Live Tables パイプラインは、ストリーミングテーブルごとに自動的に作成されます。ストリーミングテーブルは、Kafka およびクラウドオブジェクトストレージからの増分データロードに使用できます。

この記事では、ストリーミングテーブルを使用して、Unity Catalog ボリューム (推奨) または外部ロケーションとして構成されたクラウドオブジェクトストレージからデータを読み込む方法について説明します。

注：

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

始める前に

開始する前に、次の要件を満たす必要があります。

ワークスペースの要件:

サーバーレスが有効になっているDatabricks 。詳細については、「サーバーレスSQLウェアハウスの有効化」を参照してください。
Unity Catalog が有効になっているワークスペース。詳細については、「 Unity Catalog の設定と管理」を参照してください。

コンピュート要件:

次のいずれかを使用する必要があります。

Current チャンネルを使用するSQLウェアハウス。
Databricks Runtime 13.3 LTS以降の共有アクセスモードを使用したコンピュート。
Databricks Runtime 15.4 LTS 以上のシングルユーザーアクセスモードを備えたコンピュート。

Databricks Runtime 15.3 以前では、シングルユーザーコンピュートを使用して、他のユーザーが所有するストリーミングテーブルをクエリすることはできません。シングルユーザーコンピュートを Databricks Runtime 15.3以下で使用できるのは、ストリーミングテーブルを所有している場合のみです。テーブルの作成者が所有者です。

Databricks Runtime 15.4 LTS 以降では、テーブルの所有権に関係なく、シングルユーザーコンピュートで Delta Live Tables で生成されたテーブルに対するクエリがサポートされています。 15.4 以降で提供されているデータDatabricks RuntimeLTS フィルタリングを利用するには、 Live Tables で生成されたテーブルをサポートするデータフィルタリング機能がサーバレスコンピュートで実行されるため、 Deltaワークスペースがサーバレスコンピュートに対して有効になっていることを確認する必要があります。シングルユーザーコンピュートを使用してデータフィルタリング操作を実行すると、サーバレスコンピュートリソースに対して課金される場合があります。「シングルユーザーコンピュートに対するきめ細かなアクセス制御」を参照してください。

権限の要件:

Unity Catalog外部ロケーションに対する READ FILES 権限。詳細については、「クラウドストレージをDatabricksに接続するための外部ロケーションの作成」を参照してください。
ストリーミングテーブルを作成するカタログに対するUSE CATALOG権限。
ストリーミングテーブルを作成するスキーマに対するUSE SCHEMA権限。
ストリーミングテーブルを作成するスキーマに対するCREATE TABLE権限。

その他の要件:

ソースデータへのパス。

ボリューム・パスの例: /Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

外部ロケーションパスの例: s3://myBucket/analysis

注：

この記事では、ロードするデータが、アクセスできるUnity Catalogボリュームまたは外部ロケーションに対応するクラウドストレージの場所にあることを前提としています。

ソースデータの検出とプレビュー

ワークスペースのサイドバーで、 「クエリ」をクリックし、 「クエリの作成」をクリックします。
クエリエディターで、ドロップダウンリストから Current チャンネルを使用するSQLウェアハウスを選択します。
次の内容をエディターに貼り付け、ソースデータを識別する情報の代わりに山括弧 ( <> ) 内の値を使用して、 [実行]をクリックします。

注：

関数のデフォルトがデータを解析できない場合、 read_filesテーブル値関数を実行するとスキーマ推論エラーが発生する可能性があります。たとえば、複数行の CSV または JSON ファイルの場合は、複数行モードを構成する必要がある場合があります。パーサーオプションの一覧については、「テーブル値関数read_files」を参照してください。
```
/* Discover your data in a volume */
LIST "/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>"

/* Preview your data in a volume */
SELECT * FROM read_files("/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>") LIMIT 10

/* Discover your data in an external location */
LIST "s3://<bucket>/<path>/<folder>"

/* Preview your data */
SELECT * FROM read_files("s3://<bucket>/<path>/<folder>") LIMIT 10
```

ストリーミングテーブルへのデータのロード

クラウドオブジェクトストレージのデータからストリーミングテーブルを作成するには、次の内容をクエリエディターに貼り付けて、 [実行]をクリックします。

/* Load data from a volume */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>')

/* Load data from an external location */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('s3://<bucket>/<path>/<folder>')

DLT パイプラインを使用してストリーミングテーブルを更新する

このセクションでは、クエリで定義されたソースから入手可能な最新のデータを使用してストリーミングテーブルを更新するためのパターンについて説明します。

ストリーミングテーブルをCREATEまたはREFRESHすると、更新はサーバレス Delta Live Tables パイプラインを使用して処理されます。定義する各ストリーミングテーブルには、Delta Live Tables パイプラインが関連付けられています。

REFRESHコマンドを実行すると、DLT パイプラインリンクが返されます。 DLT パイプラインリンクを使用して、更新のステータスを確認できます。

注：

ストリーミングテーブルを更新して最新のデータを取得できるのは、テーブル所有者だけです。テーブルを作成したユーザーが所有者であり、所有者は変更できません。

「 Delta Live Tables とは」を参照してください。

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

デフォルトでは、 read_files関数はテーブルの作成中にソースディレクトリ内の既存のデータをすべて読み取り、更新ごとに新しく到着するレコードを処理します。

テーブルの作成時にソースディレクトリに既に存在するデータを取り込まないようにするには、 includeExistingFilesオプションをfalseに設定します。つまり、テーブルの作成後にディレクトリに到着したデータのみが処理されます。例えば：

CREATE OR REFRESH STREAMING TABLE my_bronze_table
AS SELECT *
FROM STREAM read_files('s3://mybucket/analysis/*/*/*.json', includeExistingFiles => false)

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

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

例：

REFRESH STREAMING TABLE my_bronze_table FULL

ストリーミングテーブルの自動更新をスケジュールする

定義されたスケジュールに基づいてストリーミングテーブルが自動的に更新されるように構成するには、次の内容をクエリエディターに貼り付けて、 [実行]をクリックします。

ALTER STREAMING TABLE
[[<catalog>.]<database>.]<name>
ADD [SCHEDULE [REFRESH]
        CRON '<cron-string>'
                [ AT TIME ZONE '<timezone-id>' ]];

更新スケジュールクエリの例については、「ALTER STREAMING TABLE」を参照してください。

更新の状態を追跡する

ストリーミングテーブルの更新のステータスを確認するには、 Delta Live Tables UI でストリーミングテーブルを管理するパイプラインを表示するか、ストリーミングテーブルの DESCRIBE EXTENDED コマンドによって返される更新情報を表示します。

DESCRIBE EXTENDED <table-name>

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

Kafka からのストリーミング取り込みの例については、 read_kafkaを参照してください。

ストリーミングテーブルへのアクセス権をユーザーに付与する

ユーザーにストリーミングテーブルに対するSELECT権限を付与してクエリを実行できるようにするには、次の内容をクエリエディターに貼り付けて、 [実行]をクリックします。

GRANT SELECT ON TABLE <catalog>.<schema>.<table> TO <user-or-group>

Unity Catalogのセキュリティ保護可能なオブジェクトに対する権限の付与の詳細については、 Unity Catalog権限とセキュリティ保護可能なオブジェクト」を参照してください。

Databricks SQLのストリーミング テーブルを使用してデータを読み込む