スタンドアロンパイプラインでPythonを使用する

Important

サーバーレス汎用コンピューティングのノートブックからスタンドアロンの具体化されたビューとストリーミングテーブルの作成と更新はベータ版であり、一部のリージョンで利用できます。「ノートブック」を参照してください。

Pythonを使用して、ノートブックからスタンドアロンの具体化されたビューとストリーミングテーブルを作成および更新できます。これにより、スタンドアロンパイプラインを他のPythonベースのノートブックワークフローと共に管理できます。

これを行うには、次の 2 つの方法があります。

pyspark.pipelinesデコレーター、@dp.materialized_view、@dp.tableを配置したテーブルを定義します。ロジックがDataFrameコードとして表現しやすい場合にこれを使います。「パイプラインデコレーター付きのテーブルを定義する」を参照してください。
DatabricksのSQLウェアハウスが実行するSQL文と同じものを spark.sql()に渡して提出します。これにより、スタンドアロンのマテリアライズドビューとストリーミングテーブルに関する SQL 機能一式（REFRESH ステートメントや更新スケジュールを含む）を利用できます。「SQL ステートメントを spark.sql() で提出する」を参照してください。

スタンドアロンパイプラインのPythonソースには、サーバーレス汎用コンピューティングにアタッチされたノートブックが必要です。 Python を使用して Databricks SQL ウェアハウスからスタンドアロンパイプラインを作成または更新することはできません。これは、ウェアハウスでは Python ノートブックではなく SQL ステートメントが実行されるためです。代わりに SQL ウェアハウスを使用するには、「スタンドアロンの具体化されたビューを使用する」と「スタンドアロンストリーミングテーブルを使用する」を参照してください。

Requirements

Pythonを使用してスタンドアロンパイプラインを作成および更新するには、Databricks Runtime 18.1 以降のサーバーレス汎用コンピューティングにアタッチされたノートブックが必要です。リージョンの可用性やアクセス許可など、要件の完全な一覧については、「ノートブック」を参照してください。

パイプラインデコレーターでテーブルを定義します

Lakeflowパイプラインで使うデコレーターと同じもので、スタンドアロンのマテリアライズドビューやストリーミングテーブルを定義できます。各装飾関数は1つのテーブルを定義します。セルを実行すると、Azure Databricksがテーブルを作成し、サーバーレスパイプラインでそれを埋めます。アップデートが完了するとセルが戻ります。

Warning

パイプラインデコレーターはサーバーレス環境バージョン 5以上が必要です。

具現化されたビューを定義する

バッチデータフレームを返す関数に @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 を使います。

ストリーミングテーブルを定義する

ストリーミングデータフレームを返す関数に @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")

関数がバッチデータフレームを返す場合、 @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 ジョブ」を参照してください。

テーブルの設定

デコレーターはパイプライン内で受け入れている共通のデータセットパラメータ、例えば 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 と表を参照してください。

private=True はサポートされていません。なぜなら、プライベートテーブルは同じパイプライン内の他のデータセットのみが読み取れるからです。

サポートされていない API

スタンドアロンテーブルは単一のデータセットで単一のフローを持つため、データセット間の関係を記述するAPIは利用できません。以下はパイプライン外でエラーを発生させます:

@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パイプラインを作成してください。 Python を使用した Develop パイプラインコードを参照してください。

SQL文を提出します `spark.sql()`

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 構文を含む) の使用方法については、「スタンドアロンマテリアライズドビューの使用」および「スタンドアロンストリーミングテーブルの使用」を参照してください。

制限事項

サーバーレスの一般的なコンピューティングで作成されたスタンドアロンの具体化されたビューとストリーミングテーブルには、非同期更新のサポートやテーブルごとのコスト属性なしなど、追加の制限があります。完全な一覧については、「サーバーレス汎用コンピューティング」を参照してください。

これらのパイプラインはSQLウェアハウスではなくサーバーレスの一般的な計算上で動作するため、囲み込むウェアハウスからカスタムタグを引き継ぎません。ウェアハウスタグの system.billing.usage への伝播は、SQLウェアハウスから文が実行されるマテリアライズされたビューやストリーミングテーブルにのみ適用されます。カスタムタグを使用した SQL ウェアハウスの属性コストを参照してください。

パイプラインデコレーターで定義されたテーブルには、以下の追加の制限があります:

REFRESHステートメントでリフレッシュしたり、SCHEDULEやTRIGGER ON UPDATEでリフレッシュをスケジュールすることはできません。「テーブルの更新」を参照してください。
期待、追加フロー、変更データキャプチャ(CDC)フロー、シンク、一時ビューはサポートされていません。「サポートされていないAPI」を参照してください。
private=True はサポートされません。

その他のリソース

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-08

スタンドアロン パイプラインでPythonを使用する

Requirements

パイプラインデコレーターでテーブルを定義します

具現化されたビューを定義する

ストリーミングテーブルを定義する

テーブルを更新する

テーブルの設定

サポートされていない API

SQL文を提出します spark.sql()

具体化されたビューを作成する

ストリーミング テーブルを作成する

具体化されたビューまたはストリーミング テーブルを更新する

文をパラメーター化する

その他のステートメントを実行する

制限事項

その他のリソース

フィードバック

その他のリソース

スタンドアロンパイプラインでPythonを使用する

SQL文を提出します `spark.sql()`

ストリーミングテーブルを作成する

具体化されたビューまたはストリーミングテーブルを更新する