スタンドアロンパイプラインでの Python の使用
ノートブックでPythonを使用したスタンドアロン マテリアライズドビュー と ストリーミングテーブル の作成と更新が可能です。これにより、スタンドアロン パイプラインを他の Python ベースの ノートブック ワークフローと一緒に管理できます。
これを行うには２つの方法があります：
pyspark.pipelinesデコレータ@dp.materialized_viewと@dp.tableでテーブルを定義します。ロジックを DataFrame コードとして表現する方が簡単な場合に使用します。 パイプラインデコレータでテーブルを定義するを参照してください。spark.sql()に渡すことで、Databricks SQL warehouseがランするのと同じSQLステートメントを送信します。これにより、REFRESHステートメントや更新スケジュールを含む、完全なスタンドアロン マテリアライズドビュー と ストリーミングテーブル SQL サーフェスが得られます。spark.sql()でSQLステートメントを送信を参照してください。
スタンドアロンパイプラインの Python ソースには、 Serverless 汎用コンピュート にアタッチされたノートブックが必要です。Databricks SQL WarehouseはSQLステートメントを実行し、Pythonノートブックを実行しないため、Pythonを使用してDatabricks SQL Warehouseからスタンドアロンパイプラインを作成または更新することはできません。代わりにSQL Warehouseを使用する場合は、「スタンドアロンマテリアライズドビューの使用」および「スタンドアロンストリーミングテーブルの使用」を参照してください。
要件
Pythonでスタンドアロン パイプラインを作成して更新するには、 Databricks Runtime 18.1 以降のServerless ジェネリック コンピュートにアタッチされたノートブックが必要です。リージョンの可用性や権限など、要件の完全なリストについては、「 ノートブック」を参照してください。
パイプラインデコレータを使用してテーブルを定義する
Lakeflow パイプラインで使用するのと同じデコレータを使用して、スタンドアロン マテリアライズドビュー または ストリーミングテーブル を定義できます。各デコレータ関数は、1つのテーブルを定義します。セルの実行時に、Databricksがテーブルを作成し、それを入力するServerlessパイプラインを実行します。更新が完了すると、セルが返されます。
パイプライン デコレータには、 Serverless環境のバージョン 5以上が必要です。
マテリアライズドビュー (Materialized View) の定義
バッチ DataFrame を返す関数で @dp.materialized_view を使用します。次の例では、 Wanderbricks サンプルデータセット の bookings テーブルからマテリアライズドビュー daily_booking_revenue を作成します。
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.materialized_view(name="main.default.daily_booking_revenue")
def daily_booking_revenue():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
ストリーミング読み取りからテーブルを定義するには、代わりに@dp.tableを使用します。
ストリーミングテーブルを定義する
ストリーミングDataFrameを返す関数に@dp.tableを使用します。次の例では、同じbookingsテーブルのストリーミング読み取りからストリーミングテーブルbookings_rawを作成します。
from pyspark import pipelines as dp
@dp.table(name="main.default.bookings_raw")
def bookings_raw():
return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")
関数がバッチ DataFrameを返す場合、 @dp.tableは代わりにマテリアライズドビューを作成します。唯一の例外は replace_whereで、これは常にストリーミングテーブルになります。次の例では、2025年7月1日以降のチェックインの毎日の収益を、以前の日付を再計算せずに最新の状態に保ちます：
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.table(
name="main.default.booking_revenue_rw",
replace_where=F.col("check_in") >= F.to_date(F.lit("2025-07-01")),
)
def booking_revenue_rw():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
各ランは、述語と一致する行を削除し、その範囲のみを再計算します。「REPLACE WHERE フローによるバッチ処理」を参照してください。
テーブルの更新
デコレータを使用して定義したテーブルを更新するには、それを定義するコードを再度ランします。例えば、ノートブックのセルを再ランしたり、ノートブック全体をランしたり、ノートブックをジョブとしてランしたりします。実行ごとに、テーブルが存在しない場合は作成され、存在する場合は更新されます。
ソースで使用可能なすべてのデータを再処理するには、デコレータのいずれかにfull_refresh=Trueを渡します：
@dp.table(name="main.default.bookings_raw", full_refresh=True)
def bookings_raw():
return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")
デコレータを使用して定義したテーブルでREFRESHステートメントを使用したり、SCHEDULEやTRIGGER ON UPDATEを使用して更新をスケジュールしたりすることはできません。スケジュールに基づいて更新するには、テーブルを SQL で定義するか、ノートブックをジョブとしてスケジュールします。Lakeflow Jobsを参照してください。
テーブルの構成
デコレーターは、パイプライン内部で受け入れるのと同じ共通のデータセットパラメーターを受け入れます（comment、table_properties、partition_cols、cluster_by、schema、spark_confなど）：
@dp.materialized_view(
name="main.default.daily_booking_revenue",
comment="Daily booking revenue.",
table_properties={"quality": "gold"},
cluster_by=["check_in"],
)
def daily_booking_revenue():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
パラメーターリストについては、 materialized_viewおよびtableを参照してください。
private=True はサポートされていません。これは、プライベートテーブルは同じパイプライン内の他のデータセットからしか読み取れないためです。
サポートされていない APIs
スタンドアロンテーブルは単一のフローを持つ単一のデータセットであるため、データセット間の関係を記述する APIs は利用できません。以下は、パイプラインの外部でエラーを引き起こします。
@dp.temporary_viewそしてdp.create_streaming_table@dp.append_flowおよびその他の追加フローdp.create_auto_cdc_flowそしてdp.create_auto_cdc_from_snapshot_flow@dp.replace_flowおよび REPLACE USING フローを定義するreplace_usingパラメーター。REPLACE USING フローを使用した部分的なスナップショットの置き換えを参照してください。dp.create_sink@dp.expectやなどのエクスペクテーション@dp.expect_or_fail
これらを使用するには、代わりにLakeFlow Pipelinesを作成してください。Python を使用したパイプライン コードの開発を参照してください。
SQLステートメントの送信： spark.sql()
Pythonノートブックで、 Databricks SQL warehouse から実行するのと同じステートメントを spark.sql() に渡します。スタンドアロンのマテリアライズドビューとストリーミングテーブルの構文は同じですが、ステートメントを送信する方法のみが異なります。warehouseと同様に、各CREATEまたはREFRESHステートメントは、操作を処理するServerlessパイプラインをランします。
sparkセッションはDatabricksノートブックでdefaultで利用できるため、インポートは不要です。
マテリアライズドビューの作成
次の例では、ベーステーブル base_table1 からマテリアライズドビュー mv1 を作成します。
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 の詳細については、 「マテリアライズドビューの作成」を参照してください。
ストリーミングテーブルの作成
次の例では、raw_dataテーブルからストリーミングテーブルsalesを作成します：
spark.sql("""
CREATE OR REFRESH STREAMING TABLE sales
AS SELECT product, price FROM STREAM raw_data
""")
Auto Loader でのファイルの読み込みやスケジューリングなど、 CREATE STREAMING TABLE の詳細については、「 スタンドアロン ストリーミングテーブルの使用」を参照してください。
マテリアライズドビューまたはストリーミングテーブルの更新
REFRESHステートメントを使用して、ソースからの最新のデータでスタンドアロンテーブルを更新します：
spark.sql("REFRESH MATERIALIZED VIEW mv1")
spark.sql("REFRESH STREAMING TABLE sales")
Serverless ジェネリック コンピュートでは、更新は同期的に行われます。非同期更新 (ASYNC キーワード) はサポートされていません。Serverless ジェネリック コンピュートを参照してください。
ステートメントのパラメーター化
Pythonコードから値をステートメントに渡すには、コードにハードコーディングするのではなく、SQLで名前付きパラメータマーカーを使用し、 spark.sql() の args 引数を介して値を指定します。リテラル値には、 :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構文を含む）については、 「スタンドアロンのマテリアライズドビューの使用」および「スタンドアロンのストリーミングテーブルの使用」を参照してください。
制限事項
Serverless汎用コンピュート上に作成されたスタンドアロンのマテリアライズドビューとストリーミングテーブルには、非同期更新のサポートやテーブルごとのコスト配分がないなどの追加の制限があります。一覧全体については、「 Serverless汎用コンピュート」を参照してください。
これらのパイプラインはSQL WarehouseではなくServerless汎用コンピュートで実行されるため、囲んでいるwarehouseからカスタムタグを継承しません。system.billing.usageへのwarehouseのタグ伝播は、SQL Warehouseからステートメントがランされるマテリアライズドビューとストリーミングテーブルにのみ適用されます。カスタムタグを使用してSQLウェアハウスにコストを割り当てるを参照してください。
パイプラインデコレータで定義されたテーブルには、以下の追加の制限があります：
REFRESHステートメントで更新したり、SCHEDULEやTRIGGER ON UPDATEで更新をスケジュールしたりすることはできません。テーブルの更新を参照してください。- 期待、追加フロー、チェンジデータキャプチャ（CDC）フロー、シンク、一時ビューはサポートされていません。サポートされていないAPIを参照してください。
private=Trueサポートされていません。