サーバーレスコンピュートで DataFrame をキャッシュする

Important

この機能はパブリックプレビュー段階です。ワークスペース管理者は、Previewsページからサーバーレスコンピュートでの DataFrame キャッシュへのアクセスを制御できます。 Manage Azure Databricks プレビューを参照してください。

サーバーレス計算では、df.cache() や df.persist() で DataFrame をキャッシュして、セッション内で計算コストの高い結果を再利用することができます。 Azure Databricks cache() を呼び出したときにデータフレームを計算し、その結果をスナップショットとして実体化します。以降、そのセッションが終了するまで、元の DataFrame とそこから作成した DataFrame は、再計算する代わりにスナップショットからデータを読み取ります。

Requirements

サーバーレスコンピュートでの DataFrame キャッシュ機能のプレビューは、ワークスペースで有効になっています（デフォルトで有効です）。ワークスペース管理者はPreviewsページから無効にできます。プレビューは、コンプライアンスセキュリティプロファイルが有効になっているワークスペースでは利用できません。ワークスペースレベルのプレビューの管理を参照してください。
サーバーレスコンピュート上のノートブックやジョブは、環境バージョン 5以上を使用しなければなりません。
Python版Databricks Connectは18.1.8以上、Scala版Databricks Connectは18.1.5以上でなければなりません。

古いクライアントでは、CACHE_REQUIRES_NEWER_CLIENT_ON_SERVERLESS_GC エラーが発生します。

サーバーレス計算でDataFrameキャッシュを使用するタイミング

必要なときにサーバーレスコンピュートでDataFrameをキャッシュする:

高価な結合、集約、フィルターの結果を同じセッション内で何度も再利用してください。
foreachBatch関数で一度だけストリーミングマイクロバッチを計算し、その後複数のシンクに書き込みます。
同じ準備済み特徴データ上で複数の機械学習モデルを訓練またはチューニングします。
分析の複数のステップを実行する間、たとえソーステーブルが変わっても安定したスナップショットを保ちましょう。

ベストプラクティスを参照してください。

サーバーレスコンピュートで DataFrame API をキャッシュする

以下の例は、1日分の注文を顧客データと結合してキャッシュし、その日の総収益を計算し、地域ごとの収益をレポートテーブルに追加してからキャッシュをリリースします。どちらの結果も、ジョインを再度実行する代わりにキャッシュされたスナップショットを読み込みました。

Python

from pyspark.sql import functions as F

# The DataFrame to cache: one day of orders joined with customer data.
oct_15_orders = (
    spark.table("main.sales.orders")
    .filter(F.col("order_date") == "2026-10-15")
    .join(spark.table("main.sales.customers"), "customer_id")
)
oct_15_orders.cache()

# Both results are built from oct_15_orders, so they read the cached snapshot.
total_revenue = oct_15_orders.agg(F.sum("amount")).first()[0]
print(f"Total revenue on 2026-10-15: {total_revenue}")

revenue_by_region = oct_15_orders.groupBy("order_date", "region").agg(F.sum("amount").alias("revenue"))
revenue_by_region.write.mode("append").saveAsTable("main.reports.revenue_by_region")

# Release the cache and the DataFrame when you no longer need it.
# oct_15_orders and DataFrames built from it can no longer be used.
oct_15_orders.unpersist()

Scala

import org.apache.spark.sql.functions._

// The DataFrame to cache: one day of orders joined with customer data.
val oct15Orders = spark.table("main.sales.orders")
  .filter(col("order_date") === "2026-10-15")
  .join(spark.table("main.sales.customers"), "customer_id")
oct15Orders.cache()

// Both results are built from oct15Orders, so they read the cached snapshot.
val totalRevenue = oct15Orders.agg(sum("amount")).first().get(0)
println(s"Total revenue on 2026-10-15: $totalRevenue")

val revenueByRegion = oct15Orders.groupBy("order_date", "region").agg(sum("amount").alias("revenue"))
revenueByRegion.write.mode("append").saveAsTable("main.reports.revenue_by_region")

// Release the cache and the DataFrame when you no longer need it.
// oct15Orders and DataFrames built from it can no longer be used.
oct15Orders.unpersist()

以下のAPI説明では、 df はセッション内の任意のDataFrame、例えば前述の oct_15_orders を意味します。

`df.cache()`

df を直ちに計算し、その結果をセッションの残りの期間中に使用するスナップショットとしてキャッシュします。

戻り値: キャッシュされた同じ DataFrame (df)。元のテーブルに対する計算結果ではなく、スナップショットに基づく DataFrame になります。以下の挙動をご覧ください。
動作:
- DataFrame のクエリを直ちに実行し、呼び出しから制御が戻る前に、その結果をスナップショットとして実体化します。
- 呼び出し後、 df と返されたデータフレーム(df 自体)はスナップショットに基づいています。ソーステーブルの計算にはもはや接続されていないため、クエリを再計算する代わりにスナップショットを読み込みます。
- dfから派生したデータフレームは、直接または他の派生データフレームを通じて、元のクエリを再度実行する代わりに自動的にスナップショットを読み込みます。スナップショットを読むを参照してください。
- スナップショットは、そのセッション内でのみ利用できます。 df をキャッシュしたセッションだけが、そのスナップショットを読み取れます。「隔離と特権」を参照してください。
- このスナップショットはセッション終了まで、最大7日間続きます。その後、Azure Databricksはバックグラウンドで自動的にストレージを回収します。 df.unpersist() を呼び出すと、ストレージ領域をより早く解放できるようになります。ライフサイクルと時間制限および保管およびクリーンアップを参照してください。
- もしコンピューティング df が失敗すると、df.cache() がエラーを発生させ、df はキャッシュされません。
- すでにキャッシュされているデータフレームに対して df.cache() を呼び出しても何も起こりません。 df は再び計算されません。

スナップショットは df.cache()を呼び出した後に固定されます。その後、df が読み込むテーブルに書き込まれたデータや、カラムの追加や名前変更などのスキーマ変更はそれには影響しません。最新のデータから再計算するには、ソーステーブルからDataFrameを再構築し、必要に応じて新しいDataFrameをキャッシュしてください。

スナップショットを読む

サーバーレス計算では、キャッシュマッチングはDataFrame導出に基づいています。 df.cache()後、ソーステーブルでのクエリを再計算する代わりに、以下のDataFrameのみが自動的にスナップショットを読み込みます。

キャッシュした DataFrame (df)。
select()、filter()、groupBy()、join() などを使用して df から派生させた DataFrame。
これらの派生 DataFrame を基に、さらに処理を連結して作成した DataFrame。

同じクエリを繰り返す新しいDataFrameは、ロジックは同じであってもソーステーブルを読み込みます。なぜなら、 dfから派生していないからです。

これによりキャッシュマッチングは明示的かつあなたの制御下に置かれます。データフレームは、キャッシュされたデータフレームから派生した場合にのみ、キャッシュデータを読み取ります。それ以外の DataFrame はすべて、元のテーブルからデータを読み取ります。

Python

oct_15_orders = (
    spark.table("main.sales.orders")
    .filter(F.col("order_date") == "2026-10-15")
    .join(spark.table("main.sales.customers"), "customer_id")
)
# The join runs one time, when you call cache().
oct_15_orders.cache()

# Derived from oct_15_orders: read the snapshot, so the join doesn't run again.
large_orders = oct_15_orders.filter(F.col("amount") > 1000)
revenue_by_region = oct_15_orders.groupBy("order_date", "region").agg(F.sum("amount").alias("revenue"))

# Derived from large_orders, which is derived from oct_15_orders: also reads the snapshot.
large_orders_by_region = large_orders.groupBy("region").count()

# Not derived from oct_15_orders: reads the source tables and runs the join again.
same_query = (
    spark.table("main.sales.orders")
    .filter(F.col("order_date") == "2026-10-15")
    .join(spark.table("main.sales.customers"), "customer_id")
)

Scala

val oct15Orders = spark.table("main.sales.orders")
  .filter(col("order_date") === "2026-10-15")
  .join(spark.table("main.sales.customers"), "customer_id")
// The join runs one time, when you call cache().
oct15Orders.cache()

// Derived from oct15Orders: read the snapshot, so the join doesn't run again.
val largeOrders = oct15Orders.filter(col("amount") > 1000)
val revenueByRegion = oct15Orders.groupBy("order_date", "region").agg(sum("amount").alias("revenue"))

// Derived from largeOrders, which is derived from oct15Orders: also reads the snapshot.
val largeOrdersByRegion = largeOrders.groupBy("region").count()

// Not derived from oct15Orders: reads the source tables and runs the join again.
val sameQuery = spark.table("main.sales.orders")
  .filter(col("order_date") === "2026-10-15")
  .join(spark.table("main.sales.customers"), "customer_id")

dfは作成元のデータフレームではなくスナップショットを読み取るため、それらのDataFrameを通じてdfの列を参照することはできません。制限事項を参照してください。

`df.persist(storageLevel)`

ストレージレベルを受け入れる df.cache() の別名です。

パラメーター:
- storageLevel (省略可能): StorageLevel.DISK_ONLY などの StorageLevel。サーバーレスコンピュートは任意のストレージレベルを受け入れ、それを無視します。スナップショットはすべてのストレージレベルで同じ方法で保存され、また、df.storageLevel は StorageLevel.DISK_ONLY を返します。

返却値と挙動は df.cache()と同じです。

`df.unpersist(blocking)`

キャッシュされた df スナップショットを解放し、データフレームを破棄します。

パラメーター:
- blocking (任意): 無視されます。どちらの値を指定しても、df.unpersist() は直ちに制御を返します。
戻り値: 同じ DataFrame (df)。ただし、この時点で破棄されています。
動作:
- スナップショットを解放可能な状態としてマークします。その後、Azure Databricksはキャッシュされたデータを削除し、バックグラウンドでストレージを回収します。
- df を破壊します: df およびそこから以前に派生したデータフレームは、もはや使用できず、その後それらに対してクエリを実行すると CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE が発生します。 dfを呼び出す前にdf.unpersist()から派生したDataFrameをキャッシュすると、そのDataFrameは独自のスナップショットを持っているため、使えるままです。
- 同じデータソースで作業を続けるには、キャッシュされていないDataFrameを再利用するのではなく、元のデータソースから新しいDataFrameを構築してください。
- キャッシュされていない、またはすでにアンキャッシュされたデータフレームに対して df.unpersist() を呼び出すことは no-opです。

キャッシュデータが必要なくなった場合、例えばノートブックやジョブの後続のステップでそれが読み取られなくなった場合に、df.unpersist() を呼び出してください。これによりAzure Databricksはより早くストレージを回収できます。 df.unpersist()でデータを更新しないでください。最新のデータを扱うには、ソーステーブルから新しいDataFrameを作成します。

Python

oct_15_orders = spark.table("main.sales.orders").filter(F.col("order_date") == "2026-10-15")
oct_15_orders.cache()
large_orders = oct_15_orders.filter(F.col("amount") > 1000)
orders_per_customer = oct_15_orders.groupBy("customer_id").count()
orders_per_customer.cache()  # Has its own snapshot.

# Acknowledge that the cached data is no longer needed.
# This releases the storage and destroys oct_15_orders.
oct_15_orders.unpersist()

oct_15_orders.count()  # Raises CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE.
large_orders.count()  # Raises CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE.
orders_per_customer.count()  # Reads its own snapshot.

# If you need the data again later, build a new DataFrame from the source tables.
oct_15_orders = spark.table("main.sales.orders").filter(F.col("order_date") == "2026-10-15")

Scala

var oct15Orders = spark.table("main.sales.orders").filter(col("order_date") === "2026-10-15")
oct15Orders.cache()
val largeOrders = oct15Orders.filter(col("amount") > 1000)
val ordersPerCustomer = oct15Orders.groupBy("customer_id").count()
ordersPerCustomer.cache() // Has its own snapshot.

// Acknowledge that the cached data is no longer needed.
// This releases the storage and destroys oct15Orders.
oct15Orders.unpersist()

oct15Orders.count() // Throws CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE.
largeOrders.count() // Throws CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE.
ordersPerCustomer.count() // Reads its own snapshot.

// If you need the data again later, build a new DataFrame from the source tables.
oct15Orders = spark.table("main.sales.orders").filter(col("order_date") === "2026-10-15")

`df.storageLevel`

df のストレージレベルを取得します。

戻り値: df がキャッシュされている間は StorageLevel.DISK_ONLY。キャッシュする前と df.unpersist() を呼び出した後は StorageLevel.NONE。

Python

from pyspark import StorageLevel

oct_15_orders.cache()
oct_15_orders.storageLevel == StorageLevel.DISK_ONLY  # True

Scala

import org.apache.spark.storage.StorageLevel

oct15Orders.cache()
oct15Orders.storageLevel == StorageLevel.DISK_ONLY // true

メソッドシグネチャについては、cache、persist、unpersist、およびstorageLevelに関するPySparkのリファレンスを参照してください。

ライフサイクルと時間制限

キャッシュされたデータは、それをキャッシュした Azure Databricks セッション内にのみ存在します。セッション終了時にはアクセスできなくなり、有効期間は最大7日間です。 7日間の制限を過ぎると、キャッシュされたDataFrameを使用すると、ソーステーブルを再度読み込む代わりにエラーが発生します。再度キャッシュするには、新しいセッションを開始してください。

ストレージとクリーンアップ

Azure Databricksはキャッシュデータのストレージを自動的に管理します。キャッシュされたデータはクラウドストレージに保存され、具体的なストレージ場所はワークスペースの種類によって異なります。

サーバーレスワークスペース:キャッシュされたデータはデフォルトのストレージに保存されます。マネージドサービス向けに顧客管理キーを設定している場合、キャッシュデータはそのキーを使用します。「暗号化用のカスタマーマネージドキー」を参照してください。

クラシックワークスペース:キャッシュされたデータはワークスペース作成時に設定されたストレージアカウントに保存されます。暗号化はストレージアカウントレベルで設定できます。

キャッシュされたデータにアクセスできなくなると、Azure Databricks がバックグラウンドでストレージ領域を自動的に解放します。これには、切断やクラッシュでセッションが終了する場合も含まれます。 df.unpersist()を呼び出すことで、Azure Databricksはより早くストレージを回収できます。

分離と特権

サーバーレスコンピュートでコードを実行できるユーザーは誰でもDataFrameをキャッシュできます。追加の権限は必要ありません。

キャッシュデータはセッションレベルの分離を提供します。データフレームをキャッシュしたセッションのみがそのキャッシュデータを読み取ることができます。同じノートブックの他のユーザーを含め、他のユーザーはそれにアクセスできません。

キャッシュされたデータには、キャッシュしたときに読み取ることが許可された内容が含まれています。その後、ソーステーブルに適用された権限の変更、行フィルター、カラムマスクはキャッシュされたデータには適用されません。

ベストプラクティス

安定したスナップショットが必要なときにキャッシュする:同じ計算データに対して複数の変換、集約、ドリルダウンを実行すると、スナップショットはソーステーブルが変わっても変わらない一貫したベースラインを提供します。
単純なテーブルの読み取りではなく、計算負荷の高い処理をキャッシュしてください。df.cache() はクエリ結果を実体化し、以降の読み取りではそのスナップショットをスキャンします。 spark.table("main.sales.orders").cache()のような単純なテーブルリードをキャッシュしても、後の読み取りはテーブルの完全なコピーをスキャンするため、キャッシュは節約にあまり効果がありません。キャッシュは、結合や集約、複雑なフィルターなどのコストの高い処理で効果を発揮します。
一度しか読み取らないデータをキャッシュしないでください： df.cache() は結果をすぐに計算・保存しますが、それには時間とストレージが必要です。結果を一度だけ読んだら、初期コストは割に合いません。
キャッシュされたDataFrameから導出してキャッシュを活用します:スナップショットを読み取るには、キャッシュされたDataFrame変数から後の操作(例えば df.select(...) や df.filter(...))を作成します。ソーステーブルに対して同じ計算から構築された独立したDataFrameはキャッシュを使いません。
キャッシュはクロスセッションやクロスユーザー間のデータ共有に使わないでください:キャッシュされたデータはセッション専用です。他のセッション、ノートブック、ジョブラン、そしてユーザーはそれを読めません。セッション間や他のユーザーと計算結果を共有するには、それらを恒久的なテーブルに書き込みます。
キャッシュされたデータを含むセッションを7日以内に保持してください: ライフサイクルと時間制限を参照してください。ワークフローでデータを長く保存する必要がある場合は、テーブルに書き込むのが良いでしょう。
キャッシュしたデータが不要になったら、df.unpersist() を呼び出してください。これにより、Azure Databricks がストレージ領域をより早く解放できます。呼び出し後は、DataFrameとそこから派生したDataFrameは使用できません。
データを更新する目的で df.unpersist() を使用しないでください。df.unpersist() を呼び出すと、スナップショットの有効期間が終了します。ソーステーブルの変更を検出するには、ソーステーブルから新しいDataFrameを作成してください。

制限事項

サーバーレス計算におけるDataFrameキャッシュには以下の制限があります:

列名と型:以下のDataFrameはキャッシュできません。キャッシュする前に、.alias("total_amount") などを使用して列名を変更するか、列のデータ型を変換してください。
- 重複する列名を持つDataFrames。
- 集計によって生成される sum(amount) など、列名にスペースまたは ,;{}()= のいずれかの文字が含まれる DataFrame。
- 間隔列を持つDataFrame。
- NULL型の値を含む配列、マップ、または構造体を持つDataFrame。
- 列のないDataFrame、またはNULL型のみの列。
列の参照: DataFrame をキャッシュした後は、その作成元の DataFrame を介してキャッシュ済み DataFrame の列を参照することはできません。例えば、oct_15_orders = orders.join(customers, "customer_id") と oct_15_orders.cache() の後に、oct_15_orders.select(orders.amount) が CANNOT_RESOLVE_DATAFRAME_COLUMN を発生させます。代わりに、oct_15_orders.amount タグまたは "amount" タグを使用してください。
SQLおよびカタログキャッシュAPI: CACHE TABLEなどのSQLキャッシュ文や spark.catalog のテーブルキャッシュメソッドはサーバーレスコンピュートではサポートされていません。「キャッシュの制限事項」を参照してください。 SQLでクエリ結果を再利用するには、セッションローカルの一時テーブルを使用します。
情報スキーマクエリ:メタストア内の多くのセッションがDataFrameをキャッシュしている場合、 information_schema.volumes や information_schema.schemata でのクエリが遅くなったりタイムアウトしたりすることがあります。カタログで volumes をフィルタリングしてもクエリが速くなるわけではありません。 information_schema.tablesに関するクエリは影響を受けません。

Azure Government regions:Azure Government regionsではServerless computeではDataFrameキャッシュがサポートされていません。

無視されるパラメーター: df.persist(storageLevel) はストレージレベルを無視し、df.unpersist(blocking) は blockingを無視します。「df.persist(storageLevel)」と「df.unpersist(blocking)」を参照してください。

クラシックコンピュートからの移行

ほとんどのワークロードでは、サーバーレス計算でのキャッシュはクラシック計算と同じ結果をもたらし、既存コードは変更なしで動作します。次の表に相違点を示します。

Behavior	サーバーレスコンピューティング	クラシックコンピューティング
キャッシュマッチング	キャッシュを自動的に使うのはキャッシュされたDataFrameとそこから派生したDataFrameのみです。	同じプラン形状のクエリは自動的にキャッシュを使用します。
データの鮮度	`df.cache()` を呼び出した時点で取得される整合性のあるスナップショット。	ケースによって異なります: キャッシュされたデータは場合によっては更新され、また別のケースではスナップショットとして機能します。
共有範囲	このセッション内でのみ非公開です	クラスター内のセッション間で共有されます。
具体化	即時: `df.cache()`を呼び出すとすぐに計算されます。	レイジー: DataFrame に対する最初の操作時に計算されます。
`df.unpersist()`	DataFrame を破棄し、そのスナップショットを確実に解放します。	インメモリキャッシュを解放します。後のDataFrame使用では、製品やデータソースによっては最新のデータや古いデータを取得する場合があります。

以下では、最初の2つの違いについてより詳しく説明します。

キャッシュマッチング:
- 変更で影響を受けるワークロード:ほとんどのワークロードは影響を受けません。例えば df.select(...) や df.filter(...)など、キャッシュされたDataFrame変数を使い続けると、キャッシュは以前と同じように動作します。この変更は、独立して作成されたDataFrameやSQLクエリが、同じプラン形状であるためキャッシュを再利用するワークロードにのみ影響します。サーバーレス計算では、それらのクエリはソーステーブルを読み込みます。
- もしワークロードに影響が出た場合は、同じクエリを何度も作るのではなく、キャッシュされたDataFrame変数を必要な場所で使うと良いでしょう。
データの新鮮さ:
- 変更の影響を受けるワークロード:ほとんどのワークロードは影響を受けません。なぜなら、ソーステーブルに書き込みた後にキャッシュデータを再度読み込むことは稀だからです。この変更は、ソーステーブルに書き込み、キャッシュされたDataFrameが新しいデータを表示することを期待するワークロードにのみ影響します。
- もしワークロードに影響が出た場合:ソーステーブルに書き込みた後、キャッシュされたものを再利用する代わりに、そこから新しいDataFrameを構築してください。

その他のリソース

フィードバック

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

最終更新: 2026-10-27

サーバーレスコンピュートで DataFrame をキャッシュする

Requirements

サーバーレス計算でDataFrameキャッシュを使用するタイミング

サーバーレス コンピュートで DataFrame API をキャッシュする

Python

Scala

df.cache()

スナップショットを読む

Python

Scala

df.persist(storageLevel)

df.unpersist(blocking)

Python

Scala

df.storageLevel

Python

Scala

ライフサイクルと時間制限

ストレージとクリーンアップ

分離と特権

ベスト プラクティス

制限事項

クラシックコンピュートからの移行

その他のリソース

フィードバック

その他のリソース

サーバーレスコンピュートで DataFrame API をキャッシュする

`df.cache()`

`df.persist(storageLevel)`

`df.unpersist(blocking)`

`df.storageLevel`

ベストプラクティス