在无服务器计算上缓存数据帧

Important

此功能目前以公共预览版提供。工作区管理员可以通过预览页面控制对无服务器计算上的 DataFrame 缓存的访问。请参阅 Manage Azure Databricks 预览版。

在无服务器计算中，你可以使用 df.cache() 或 df.persist() 缓存 DataFrame，以便在会话中重用代价高昂的结果。 Azure Databricks 在你调用cache()时计算 DataFrame，并将结果物化为快照。 DataFrame 以及你基于它构建的数据帧会在本次会话的剩余时间内读取该快照，而不是重新计算它。

Requirements

您的工作区已启用无服务器计算上的 DataFrame 缓存预览版（默认启用）。 Workspace 管理员可以在预览页面禁用它。在启用合规安全配置文件的工作区中，预览版无法使用。请参阅管理工作区级预览。
无服务器计算上的笔记本和作业必须使用环境版本 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

以下示例缓存一天的订单与客户数据合并，计算当天的总收入，按地区将收入附加到报告表中，然后释放缓存。这两个结果都是读取缓存快照，而不是重新运行联接。

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，并将结果缓存为在本次会话的剩余时间内使用的快照。

返回：相同的缓存数据帧， df现在基于快照而非源表的计算。请参见以下现象。
行为：
- 立即运行 DataFrame 的查询，并在调用返回前将结果物化为快照。
- 调用后，df 和返回的数据帧（df 本身）都基于快照。它们不再与源表上的计算相关联，因此会读取快照而不是重新计算查询。
- 直接或通过其他派生数据帧派生的df数据帧会自动读取快照，而不是再次运行原始查询。参见查看快照。
- 该快照仅限于当前会话。只有缓存 df 的会话能读取它。参见隔离与特权。
- 快照持续到会话结束，最多七天。 Azure Databricks 随后会自动在后台回收其存储空间。调用 df.unpersist() 可以让它更早地回收存储空间。参见生命周期和时间限制以及存储与清理。
- 如果计算 df 失败，df.cache() 会引发该错误，且 df 不会被缓存。
- 对已缓存的 DataFrame 调用 df.cache() 不会有任何作用。 df 不会再次计算。

在你调用 df.cache()后，快照会被固定。之后写入到 df 读取的表中的数据，以及对这些表的模式修改（例如添加或重命名列），都不会影响它。要从最新数据重新计算，可以从源表重新构建DataFrame，如果需要可以缓存新的DataFrame。

阅读快照

在无服务器计算中，缓存匹配基于 DataFrame 推导。之后 df.cache()，只有以下数据帧自动读取快照，而无需在源表上重新计算查询：

你缓存的数据帧是 df。
由 df派生的数据帧，例如，带有 select()、 filter()、 groupBy()或 join()。
从这些衍生数据帧进一步链式派生的 DataFrame。

一个重复相同查询的新DataFrame会读取源表，尽管其逻辑相同，因为它并非从 df 派生而来。

这使得缓存匹配变得显式且由你控制：只有当一个 DataFrame 是从已缓存的 DataFrame 推导而来时，它才会读取缓存数据。每两个 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 读取的是快照而不是构建它的数据帧，所以你无法通过这些数据帧引用 df 的列。请参阅限制。

`df.persist(storageLevel)`

接受存储级别的 df.cache() 别名。

参数：
- storageLevel （可选）：StorageLevel，例如 StorageLevel.DISK_ONLY。无服务器计算支持任何存储层级并忽略它。快照对每个存储层的存储方式相同，并 df.storageLevel 返回 StorageLevel.DISK_ONLY。

返回值和行为与 df.cache()相同。

`df.unpersist(blocking)`

释放 df 的缓存快照，并销毁 DataFrame。

参数：
- blocking （可选）：将被忽略。 df.unpersist() 无论哪种值，都能立即返回。
返回：同一数据帧， df现已销毁。
行为:
- 将快照标记为可回收。 Azure Databricks 随后会删除缓存数据，并在后台回收存储空间。
- 销毁df：df以及之前从中导出的任何数据帧都不再可用，后续对其的任何查询都会引发CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE。如果你在调用df之前缓存了从df.unpersist()衍生出来的数据帧，该数据帧依然可用，因为它有自己的快照。
- 要继续使用同一数据源，请从原始数据源构建新的DataFrame，而不是重复使用未缓存的数据帧。
- 对从未缓存或已取消缓存的 DataFrame 调用 df.unpersist() 不会产生任何效果。

当你不再需要缓存数据时，比如笔记本或作业中没有后续步骤读取缓存时，再调用 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

关于方法签名，请参见 PySpark 中关于 cache、persist、unpersist 和 storageLevel 的参考文档。

生命周期和时间限制

缓存数据只存在于你缓存的 Azure Databricks 会话中。会话结束后，它将变得无法访问，持续时间最多为七天。超过七天限制后，使用缓存的 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(...)。基于源表相同计算构建的独立数据帧不使用缓存。
不要用缓存进行跨会话或跨用户数据共享：缓存数据对你的会话是私有的。其他会话、笔记本、作业运行和用户都无法读取它。为了在会话间或与其他用户共享计算结果，可以写入永久表。
保持缓存数据会话在七天以内：参见生命周期和时间限制。如果你的工作流程需要将数据保留更长时间，就将其写入表中。
当不再需要缓存数据时调用df.unpersist()：这允许 Azure Databricks 更早回收存储。调用后，DataFrame及其衍生的数据帧无法使用。
不要用 df.unpersist() 来刷新数据：使用 df.unpersist() 会结束该快照的生命周期。要获取源表中的变更，可以从源表中构建一个新的DataFrame。

Limitations

无服务器计算中的DataFrame缓存存在以下限制：

列名和类型：以下数据帧无法缓存。在缓存前重命名或转换列，例如使用 .alias("total_amount")。
- 带有重复列名的数据帧。
- 列名包含空格或字符 ,;{}()= 中任意字符的数据帧，例如聚合产生的 sum(amount)。
- 带有区间列的数据帧。
- 包含NULL类型值的数组、映射或结构体的数据帧。
- 没有列的DataFrame，或仅有NULL类型的列。
列引用：缓存 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：SQL 缓存语句（例如 CACHE TABLE）以及 spark.catalog 的表缓存方法在无服务器计算中不受支持。参见缓存限制。要在 SQL 中重用查询结果，可以使用会话本地的临时表。
信息模式查询：当元存储库中多个会话缓存数据帧时，查询information_schema.volumesinformation_schema.schemata可能会变慢或超时。按目录过滤volumes并不会让查询更快。查询 information_schema.tables 不会受到影响。

Azure 政府区域：在 Azure 政府区域的无服务器计算中不支持 DataFrame 缓存。

忽略参数： df.persist(storageLevel) 忽略存储层级， df.unpersist(blocking) 忽略 blocking。请参阅 df.persist(storageLevel) 和 df.unpersist(blocking)。

从经典计算环境迁移

对于大多数工作负载，在无服务器计算上缓存的结果与经典计算相同，现有代码无需更改即可运行。下表列出了差异。

Behavior	无服务器计算	经典计算
缓存匹配	只有缓存中的 DataFrame 及其衍生的数据帧会自动使用缓存。	任何具有相同计划形状的查询都会自动使用缓存。
数据新鲜度	在你呼叫 `df.cache()` 时拍摄的一致性快照。	情况各异：缓存数据在某些情况下会被刷新，在另一些情况下则提供快照。
共享范围	仅当前会话可见。	集群内的会话间共享。
物化	立即求值：当你调用 `df.cache()` 时立即计算。	懒惰：在首次对 DataFrame 执行操作时计算。
`df.unpersist()`	销毁 DataFrame 并以确定性方式释放其快照。	释放内存缓存。后续使用DataFrame时，根据产品和数据源的不同，可能会获得最新或过时的数据。

以下更详细地描述了前两个差异：

缓存匹配：
- 受变更影响的工作负载：大多数工作负载不受影响。如果你继续使用缓存的 DataFrame 变量，例如，使用 df.select(...) 或 df.filter(...)，缓存就会像以前一样工作。该变更只影响那些独立构建的 DataFrame 或 SQL 查询因相同计划形状而重复使用缓存的工作负载。在无服务器计算中，这些查询读取源表。
- 如果你的工作负载受到影响：请在需要缓存结果的地方使用缓存的 DataFrame 变量，而不是重复构建同样的查询。
数据新鲜度：
- 受变更影响的工作负载：大多数工作负载不受影响，因为写入源表后重新读取缓存数据的情况较少。该更改只影响那些写入源表且期望缓存数据帧显示新数据的工作负载。
- 如果你的工作负载受到影响：写入源表后，从它们构建新的 DataFrame，而不是重复使用缓存的 DataFrame。

其他资源

反馈

此页面是否有帮助？

上次更新时间为 2026-10-23