メインコンテンツまでスキップ
に最終更新

Serverless コンピュート上のキャッシュ DataFrames

備考

プレビュー

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

Serverlessコンピュートでは、df.cache() または df.persist() で DataFrame をキャッシュして、セッション内の高コストな計算結果を再利用できます。cache() を呼び出すと、Databricks は DataFrame をコンピュートし、その結果をスナップショットとしてマテリアライズします。その データフレーム とそれに基づいて構築された DataFrames は、残りのセッションでスナップショットを再計算する代わりにスナップショットを読み取ります。

要件​

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

古いクライアントはCACHE_REQUIRES_NEWER_CLIENT_ON_SERVERLESS_GCエラーを受け取ります。

Serverless コンピュートで DataFrame キャッシュを使用する場合​

Serverless コンピュートで DataFrame をキャッシュする必要がある場合：

  • 高コストな結合、集計、またはフィルターの結果を同じセッションで複数回再利用します。
  • foreachBatch 関数でストリーミングマイクロバッチを 1 回コンピュートし、複数のシンクに書き込みます。
  • 同一の準備済み特徴量データで、複数の機械学習モデルのトレーニングまたはチューニングを行います。
  • ソーステーブルが変更された場合でも、分析の複数のステップを実行している間は安定したスナップショットを保持します。

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

Serverless コンピュートで DataFrame APIs をキャッシュする​

次の例では、顧客データと結合された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()

次の API 説明では、 df は、前の例の oct_15_orders など、セッション内のすべての DataFrame を表します。

df.cache()​

dfを即座にコンピュートし、セッションの残りの期間、結果を スナップショット としてキャッシュします。

  • 戻り値 ：同じキャッシュ済み DataFrame df。現在は、ソース テーブルに対する計算ではなく、スナップショットに基づいています。以下の動作を確認してください。
  • 動作 :
    • DataFrame のクエリーをすぐに実行し、呼び出しが戻る前に結果をスナップショットとしてマテリアライズします。
    • 呼び出しの後、df と返された DataFrame (df 自体) はスナップショットに基づいています。それらはソーステーブルでの計算に接続されなくなったため、クエリーを再計算するのではなくスナップショットを読み取ります。
    • df から直接、または他の派生 DataFrames を介して派生した DataFrames は、元のクエリーを再度実行する代わりにスナップショットを自動的に読み取ります。「スナップショットの読み取り」を参照してください。
    • スナップショットはセッションに対してローカルです。dfをキャッシュしたセッションのみがそれを読み取ることができます。分離と権限を参照してください。
    • スナップショットは、最大7日間、セッションが終了するまで保持されます。その後、Databricks はバックグラウンドで自動的にストレージを再利用します。df.unpersist()を呼び出すと、ストレージを早期に再利用できます。ライフサイクルと時間制限およびストレージとクリーンアップを参照してください。
    • dfの計算が失敗した場合、df.cache()はエラーを発生させ、dfはキャッシュされません。
    • すでにキャッシュされているDataFrameに対してdf.cache()を呼び出しても、何も起こりません。dfは再コンピュートされません。

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

スナップショットを読み取る​

Serverlessコンピュートでは、キャッシュマッチングは DataFrame の派生に基づいています。df.cache()の後、ソーステーブルでのクエリーの再計算ではなく、次のDataFrameのみが自動的にスナップショットを読み取ります：

  • キャッシュした DataFrame 、 df。
  • df から派生した DataFrames 、たとえば select()、 filter()、 groupBy()、または join()。
  • これらの派生DataFramesからさらに連鎖したDataFrames。

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

これにより、キャッシュマッチングが明示的になり、ユーザーが制御できるようになります。データフレームは、キャッシュされた データフレームから派生した場合にのみ、キャッシュされたデータを読み取ります。その他のすべての データフレームは、ソーステーブルを読み取ります。

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")
)

df は、それが構築された DataFrames ではなくスナップショットを読み取るため、これらの DataFrames を介して df の列を参照することはできません。詳細は制限事項を参照してください。

df.persist(storageLevel)​

ストレージレベルを受け取るdf.cache()のエイリアス。

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

戻り値と動作は df.cache() と同じです。

df.unpersist(blocking)​

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

  • パラメーター :

    • blocking （オプション）：無視されます。df.unpersist() は、どちらの値に対しても直ちに返されます。

  • 戻り値 ：同じ DataFrame 、df、現在は破棄されています。

  • 動作 :

    • スナップショットを「回収可能」とマークします。次に、Databricksはキャッシュされたデータを削除し、バックグラウンドでストレージを回収します。
    • df を破棄します。 df と、以前にそれから派生した DataFrames は使用できなくなり、それらに対する以降のクエリーは CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE を発生させます。df.unpersist() を呼び出す前に df から派生した DataFrame をキャッシュすると、独自のスナップショットを持つため、その DataFrame は引き続き使用できます。
    • 同じデータソースでの作業を続行するには、キャッシュされていない DataFrame を再利用するのではなく、元のデータソースから新しい DataFrame を構築します。
    • キャッシュされていない、または既にキャッシュが解除されているDataFrameでdf.unpersist()を呼び出しても、何も起こりません。

ノートブックやジョブの後のステップでキャッシュデータが読み取られない場合など、キャッシュされたデータが不要になったら df.unpersist() を呼び出します。これにより、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")

df.storageLevel​

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

  • 戻り値 : StorageLevel.DISK_ONLY、 df がキャッシュされている場合。StorageLevel.NONE をキャッシュし、 df.unpersist() を呼び出した後。
Python
from pyspark import StorageLevel

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

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

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

キャッシュされたデータは、キャッシュしたDatabricksセッション内でのみ存在します。セッションが終了するとアクセスできなくなり、最大7日間保持されます。7日間の制限を過ぎると、キャッシュされたDataFrameを使用すると、ソーステーブルを再読み込みするのではなく、エラーが発生します。もう一度キャッシュするには、新しいセッションを起動します。

保管とクリーンアップ​

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

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

  • クラシックワークスペース : キャッシュされたデータは、ワークスペース作成時に構成されたワークスペースストレージバケットに保存されます。ストレージバケットレベルで暗号化を構成できます。

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

分離と権限​

Serverless コンピュートでコードを実行できるユーザーであれば、 DataFrame をキャッシュできます。追加の権限は必要ありません。

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

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

ベストプラクティス​

  • 安定したスナップショットが必要な場合のキャッシュ : 同じコンピュート・データに対して複数の変換、集計、またはドリルダウンをランする場合、スナップショットを使用すると、ソース・テーブルが変更されても変更されない一貫したベースラインが提供されます。
  • 単純なテーブルの読み取りではなく、計算コストの高い計算をキャッシュする : df.cache() はクエリー結果をマテリアライズし、後の読み取りはそのスナップショットをスキャンします。spark.table("main.sales.orders").cache() のような単純なテーブルの読み取りをキャッシュする場合、後の読み取りはテーブルの完全なコピーをスキャンするため、キャッシュによる節約はほとんどありません。キャッシュは、結合、集約、複雑なフィルタなど、コストのかかる作業の後に効果を発揮します。
  • 1回しか読み取らないデータをキャッシュしない : df.cache()は結果をすぐにコンピュートして保存するため、時間とストレージが必要になります。結果を1回しか読み取らない場合、初期コストは意味がありません。
  • キャッシュされた DataFrame から派生してキャッシュを使用する : スナップショット を読み取るには、 df.select(...) や df.filter(...) など、キャッシュされた DataFrame 変数から後続の操作を構築します。ソース テーブルで同じ計算から構築された独立した DataFrame は、キャッシュを使用しません。
  • セッション間またはユーザー間のデータ共有にキャッシュを使用しないでください ：キャッシュされたデータはセッション専用です。他のセッション、ノートブック、ジョブ実行、ユーザーは読み取ることができません。コンピュートされた結果をセッション間で、または他のユーザーと共有するには、それらを永続テーブルに書き込みます。
  • キャッシュされたデータを持つセッションを7日未満に保持する :ライフサイクルと時間制限を参照してください。ワークフローでより長い時間データが必要な場合は、データをテーブルに書き込んでください。
  • キャッシュされたデータが不要になったら df.unpersist() を呼び出す : これにより、Databricks はストレージを早期に再利用できます。呼び出し後、その DataFrame と、そこから派生した DataFrames は使用できなくなります。
  • df.unpersist()を使用してデータを更新しないでください ：df.unpersist()によってスナップショットの有効期限が終了します。ソーステーブルの変更を反映するには、ソーステーブルから新しい DataFrame を構築します。

制限​

Serverless コンピュートでの DataFrame キャッシュには、次の制限があります。

  • 列の名前と型 : 以下の DataFrame はキャッシュできません。キャッシュする前に列の名前を変更するか、キャストします（例：.alias("total_amount")を使用）。

    • 列名が重複しているDataFrames。
    • 列名にスペースまたは集計の sum(amount) などの文字 ,;{}()= が含まれる DataFrames。
    • 間隔列を持つDataFrames。
    • NULL型値を含む配列、マップ、または構造体を持つDataFrames。
    • 列がないDataFrames、またはNULL型の列のみを含むDataFrames。

  • 列の参照 : DataFrame をキャッシュした後は、それが構築された DataFrames を通じてその列を参照することはできません。たとえば、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 およびカタログキャッシュ APIs : CACHE TABLEなどの SQL キャッシュ ステートメント、および spark.catalogのテーブル キャッシュ メソッドは、Serverless コンピュートではサポートされていません。キャッシュの制限事項を参照してください。SQL でクエリー結果を再利用するには、セッションローカル一時テーブルを使用します。

  • 情報スキーマ クエリー : メタストア内の多くのセッションが DataFrames をキャッシュする場合、 information_schema.volumes および information_schema.schemata のクエリーは低速になるか、タイムアウトする可能性があります。カタログによる volumes のフィルタリングは、クエリーを高速化しません。information_schema.tables のクエリーは影響を受けません。

  • AWS GovCloud : AWS GovCloud または AWS GovCloud DoD のServerless コンピュートでは、DataFrame キャッシングはサポートされていません。利用できない機能を参照してください。

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

Classic コンピュートから移行する​

ほとんどのワークロードで、Serverless コンピュートでのキャッシュは従来のコンピュートと同じ結果をもたらし、既存のコードは変更なしで実行されます。次の表に違いを示します。

挙動

サーバーレスコンピュート

Classic Compute

キャッシュマッチング

キャッシュされた DataFrame と、そこから派生した DataFrames のみが、自動的にキャッシュを使用します。

同じプラン形状のクエリーは自動的にキャッシュを使用します。

データの鮮度

df.cache() を呼び出した時に取得される一貫性のあるスナップショット。

変動します。キャッシュされたデータは特定のケースで更新され、他のケースではスナップショットを提供します。

共有のスコープ

セッションに非公開。

クラスター内のセッション全体で共有されます。

マテリアライゼーション

Eager: df.cache()を呼び出したときにすぐにコンピュートされます。

遅延: DataFrame上の最初のアクションでコンピュートされます。

df.unpersist()

DataFrame を破棄し、そのスナップショットを決定論的に解放します。

メモリ内キャッシュを解放します。DataFrame の後の使用は、製品およびデータソースに応じて、最新または古いデータを取得する場合があります。

挙動

サーバーレスコンピュート

Classic Compute

キャッシュマッチング

キャッシュされた DataFrame と、そこから派生した DataFrames のみが、自動的にキャッシュを使用します。

同じプラン形状のクエリーは自動的にキャッシュを使用します。

データの鮮度

df.cache() を呼び出した時に取得される一貫性のあるスナップショット。

変動します。キャッシュされたデータは特定のケースで更新され、他のケースではスナップショットを提供します。

共有のスコープ

セッションに非公開。

クラスター内のセッション全体で共有されます。

マテリアライゼーション

Eager: df.cache()を呼び出したときにすぐにコンピュートされます。

遅延: DataFrame上の最初のアクションでコンピュートされます。

df.unpersist()

DataFrame を破棄し、そのスナップショットを決定論的に解放します。

メモリ内キャッシュを解放します。DataFrame の後の使用は、製品およびデータソースに応じて、最新または古いデータを取得する場合があります。

次は、最初の 2 つの違いをより詳細に説明します。

  • キャッシュの一致 :

    • 変更の影響を受けるワークロード : ほとんどのワークロードは影響を受けません。キャッシュされた DataFrame 変数を引き続き使用する場合 (たとえば df.select(...) や df.filter(...)を使用する場合)、キャッシュは以前と同様に機能します。この変更は、独立して構築された DataFrame または SQL クエリーが同じプランの形状を持っているためにキャッシュを再利用するワークロードにのみ影響します。Serverless コンピュートでは、これらのクエリーはソース テーブルを読み取ります。
    • ワークロードが影響を受ける場合 : 同じクエリーを再度構築する代わりに、キャッシュされた結果が必要な場所でキャッシュされた DataFrame 変数を使用します。

  • データの鮮度 ：

    • 変更の影響を受けるワークロード : ソーステーブルに書き込んだ後にキャッシュされたDataFrameを再度読み取って新しいデータを表示することは一般的ではないため、ほとんどのワークロードは影響を受けません。この変更は、ソーステーブルに書き込み、キャッシュされたDataFrameが新しいデータを表示することを期待するワークロードにのみ影響します。
    • ワークロードが影響を受ける場合 : ソース テーブルに書き込んだ後、キャッシュされたDataFrame を再利用するのではなく、ソース テーブルから新しい DataFrame を構築します。

その他のリソース​

このページの見出し