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

/shared/preview.md

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

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

注：

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

重要

Databricks SQL で作成されたストリーミングテーブルは、サーバレス Delta Live Tables パイプラインによってサポートされます。この機能を使用するには、ワークスペースがサーバレスパイプラインをサポートしている必要があります。

始める前に

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

ワークスペースの要件:

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

コンピュート要件:

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

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

権限の要件:

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

その他の要件:

ソースデータへのパス。

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

外部ロケーションパスの例: gs://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 "gs://<bucket>/<path>/<folder>"

/* Preview your data in an external location */
SELECT * FROM read_files("gs://<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('gs://<bucket>/<path>/<folder>')

ランタイムチャンネルを設定する

SQLウェアハウスを使用して作成されたストリーミングテーブルは、Delta Live Tables パイプラインを使用して自動的に更新されます。Delta Live Tables パイプラインは、current チャンネルの by デフォルトのランタイムを使用します。リリースプロセスの詳細については、Delta Live Tables リリースノートとリリースアップグレードプロセスを参照してください。

Databricks 本番運用ワークロードには、current チャンネルを使用することをお勧めします。新機能は最初に preview チャンネルにリリースされます。プレビュー Delta Live Tables チャンネルにパイプラインを設定して、preview をテーブルプロパティとして指定することで、新機能をテストできます。このプロパティは、テーブルを作成するとき、または ALTER ステートメントを使用してテーブルを作成した後に指定できます。

次のコード例は、CREATE ステートメントでチャンネルをプレビューに設定する方法を示しています。

CREATE OR REPLACE MATERIALIZED VIEW foo.default.bar
TBLPROPERTIES ('pipelines.channel' = 'preview') as
SELECT
  *
FROM
  range(5)

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('gs://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権限とセキュリティ保護可能なオブジェクト」を参照してください。

クエリ履歴を使用した実行の監視

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

プレビュー

この機能はパブリックプレビュー段階です。ワークスペース管理者は、プレビューページからこの機能を有効にできます。「Databricksプレビューの管理」を参照してください。

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

クエリ履歴 UI で REFRESH ステートメントにアクセスするには、次の手順を使用します。

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

また、 SQL エディターのリンクを使用して、または SQLウェアハウスに添付されたノートブックからクエリの詳細にアクセスすることもできます。

注：

ストリーミングテーブルは、 preview チャンネルを使用して実行するように設定する必要があります。「ランタイムチャンネルを設定する」を参照してください。

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