Pythonを使用して、ノートブックからスタンドアロンの具体化されたビューとストリーミング テーブルを作成および更新できます。 Python ノートブックでパイプラインを作成し、spark.sql()で実行します。 これにより、スタンドアロン パイプラインを他のPythonベースのノートブック ワークフローと共に管理できます。
スタンドアロン パイプラインのPythonソースには、サーバーレス汎用コンピューティングにアタッチされたノートブックが必要です。 Python を使用して Databricks SQL ウェアハウスからスタンドアロン パイプラインを作成または更新することはできません。これは、ウェアハウスでは Python ノートブックではなく SQL ステートメントが実行されるためです。 代わりに SQL ウェアハウスを使用するには、「 スタンドアロンの具体化されたビューを使用 する」と「 スタンドアロン ストリーミング テーブルを使用する」を参照してください。
Important
サーバーレス汎用コンピューティングのノートブックからスタンドアロンの具体化されたビューとストリーミング テーブルの作成と更新は ベータ版 であり、一部のリージョンで利用できます。 「ノートブック」を参照してください。
Requirements
Pythonを使用してスタンドアロン パイプラインを作成および更新するには、Databricks Runtime 18.1 以降のサーバーレス汎用コンピューティングにアタッチされたノートブックが必要です。 リージョンの可用性やアクセス許可など、要件の完全な一覧については、「 ノートブック」を参照してください。
どのように機能するのか
Python ノートブックで、Databricks SQL ウェアハウスから実行するのと同じステートメントをspark.sql()に渡します。 スタンドアロンの具体化されたビューとストリーミング テーブルの構文は同じです。ステートメントを送信する方法のみが異なります。 ウェアハウスと同様に、各 CREATE または REFRESH ステートメントは、サーバーレス パイプラインを実行して操作を処理します。
spark セッションは既定でAzure Databricksノートブックで使用できるため、インポートは必要ありません。
具体化されたビューを作成する
次の例では、基本テーブル mv1 からマテリアライズド ビュー base_table1 を作成します。
spark.sql("""
CREATE OR REPLACE MATERIALIZED VIEW mv1
AS SELECT
date,
sum(sales) AS sum_of_sales
FROM base_table1
GROUP BY date
""")
スケジュールされた更新やトリガーされた更新など、完全な CREATE MATERIALIZED VIEW の詳細については、「 具体化されたビューを作成する」を参照してください。
ストリーミング テーブルを作成する
次の例では、sales テーブルからストリーミング テーブル raw_dataを作成します。
spark.sql("""
CREATE OR REFRESH STREAMING TABLE sales
AS SELECT product, price FROM STREAM raw_data
""")
自動ローダーを使用したファイルの読み込みとスケジュールなど、完全な CREATE STREAMING TABLE の詳細については、「 スタンドアロン ストリーミング テーブルの使用」を参照してください。
具体化されたビューまたはストリーミング テーブルを更新する
REFRESH ステートメントを使用して、スタンドアロン テーブルをソースの最新のデータで更新します。
spark.sql("REFRESH MATERIALIZED VIEW mv1")
spark.sql("REFRESH STREAMING TABLE sales")
サーバーレス汎用コンピューティングでは、更新は同期的です。 非同期更新 ( ASYNC キーワード) はサポートされていません。
サーバーレスの一般的なコンピューティングを参照してください。
文をパラメーター化する
ハードコーディングするのではなく、Python コードからステートメントに値を渡すには、SQL で名前付きパラメーター マーカーを使用し、argsのspark.sql()引数を使用してその値を指定します。 リテラル値に直接 :min_sales などのマーカーを使用します。 識別子をプレーン文字列値として置き換えることができないため、パラメーターがオブジェクト名 (テーブル、ビュー、スキーマなど) である場合にのみ、マーカーを IDENTIFIER() でラップします。
次の例では、具体化されたビュー名とフィルター値の両方をパラメーター化します。
mv_name = "main.sales.regional_sales"
min_sales = 1000
spark.sql("""
CREATE OR REPLACE MATERIALIZED VIEW IDENTIFIER(:mv)
AS SELECT
region,
sum(sales) AS sum_of_sales
FROM base_table1
WHERE sales > :min_sales
GROUP BY region
""", args={
"mv": mv_name,
"min_sales": min_sales,
})
詳細については、「 パラメーター マーカー と IDENTIFIER 句」を参照してください。
その他のステートメントを実行する
Python ノートブックからスタンドアロンの具体化されたビューまたはストリーミング テーブル ステートメントを実行するには、更新のスケジュール設定、テーブルの変更、またはテーブルの削除を行うステートメントを含む、spark.sql()に渡します。 具体化されたビューとストリーミング テーブル (SQL 構文を含む) の使用方法については、「 スタンドアロンマテリアライズド ビューの使用 」および「 スタンドアロン ストリーミング テーブルの使用」を参照してください。
制限事項
サーバーレスの一般的なコンピューティングで作成されたスタンドアロンの具体化されたビューとストリーミング テーブルには、非同期更新のサポートやテーブルごとのコスト属性なしなど、追加の制限があります。 完全な一覧については、「 サーバーレス汎用コンピューティング」を参照してください。