Cache DataFrames на бессерверных вычислениях

Important

Эта функция доступна в общедоступной предварительной версии. Администраторы рабочих пространств могут управлять доступом к кэшированию DataFrame на бессерверных вычислениях со страницы предварительного просмотра. См. статью "Управление предварительными версиями Azure Databricks".

В бессерверных вычислениях вы можете кэшировать DataFrame с помощью df.cache() или df.persist(), чтобы повторно использовать дорогостоящий результат в рамках сеанса. Azure Databricks вычисляет DataFrame при вызове cache() и материализует результат в виде снимка. DataFrame и DataFrames, которые вы на его основе создаёте, затем используют снимок на протяжении оставшейся части сеанса вместо того, чтобы пересчитывать его.

Requirements

Устаревшие клиенты получают ошибку CACHE_REQUIRES_NEWER_CLIENT_ON_SERVERLESS_GC.

Когда использовать кэширование DataFrame при бессерверных вычислениях

Кэшируйте DataFrame на бессерверных вычислительных ресурсах, когда нужно:

  • Повторно используйте результат дорогостоящего объединения, агрегации или фильтра несколько раз за одну сессию.
  • Вычислите потоковый микропакет один раз в функции foreachBatch, затем запишите его в несколько приемников.
  • Обучить или настроить несколько моделей машинного обучения на основе одних и тех же подготовленных признаков.
  • Сохраняйте стабильный снимок во время выполнения нескольких этапов анализа, даже если исходные таблицы меняются.

См. Лучшие практики.

Кэшировать API DataFrame в бессерверной вычислительной среде

Следующий пример кэширует один день заказов с данными клиента, вычисляет общий доход за день, добавляет доход по регионам в таблицу отчетов и затем освобождает кэш. В обоих случаях используется кэшированный снимок вместо повторного выполнения объединения.

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 сразу и материализует результат в виде снимка до завершения вызова.
    • После вызова df и возвращаемый DataFrame (то есть сам df) основаны на снимке. Они больше не связаны с вычислением по исходным таблицам, поэтому считывают снимок вместо повторного вычисления запроса.
    • DataFrames, полученные из df, напрямую или через другие производные DataFrames, автоматически считывают снимок вместо повторного запуска исходного запроса. См. Прочитайте моментальный снимок.
    • Снимок доступен только в рамках вашей сессии. Только сессия, которая кэшировала df , может её прочитать. См. Изоляция и привилегии.
    • Снимок длится до окончания сессии, максимум семь дней. Azure Databricks затем автоматически освобождает своё хранилище в фоновом режиме. Вызов df.unpersist() позволяет раньше освободить место в хранилище. См. Жизненный цикл и временные ограничения и Хранение и очистка.
    • Если вычисление df завершается с ошибкой, df.cache() вызывает ошибку, и df не кэшируется.
    • Вызов df.cache() для DataFrame, который уже кэширован, ничего не делает. df не вычисляется снова.

Снимок фиксируется после того, как вы вызываете df.cache(). Данные, записываемые позже в таблицы, из которых df читает данные, и изменения схемы в этих таблицах, такие как добавленные или переименованные столбцы, не влияют на df. Чтобы пересчитать на основе последних данных, постройте DataFrame заново из исходных таблиц и кэшируйте новый DataFrame, если нужно.

Прочитайте снимок состояния

В бессерверных вычислениях сопоставление кэша основано на деривании DataFrame. После df.cache() только следующие DataFrames автоматически считывают снимок вместо повторного вычисления запроса по исходным таблицам:

  • DataFrame, который вы кэшировали, df.
  • DataFrames, полученные из df, например с select(), filter(), groupBy(), или join().
  • DataFrames, далее выстроенные в цепочку из этих производных DataFrames.

Новый 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 читает снимок, а не DataFrames, из которых он был построен, нельзя ссылаться на столбцы df через эти DataFrames. См. Ограничения.

df.persist(storageLevel)

Псевдоним для df.cache(), который принимает уровень хранения.

  • Параметры:
    • storageLevel (по желанию): StorageLevel, например StorageLevel.DISK_ONLY. Serverless Compute принимает любой уровень хранения и игнорирует его. Снимок сохраняется одинаково для каждого уровня хранения и df.storageLevel возвращает StorageLevel.DISK_ONLY.

Возвратное значение и поведение совпадают с df.cache().

df.unpersist(blocking)

Освобождает кэшированный снимок df и уничтожает DataFrame.

  • Параметры:
    • blocking (по желанию): Игнорируется. df.unpersist() возвращает сразу для любого из двух значений.
  • Возвращаемое значение: тот же DataFrame, dfтеперь уничтожен.
  • Поведение:
    • Отмечает снимок как готовый к освобождению. Azure Databricks удаляет кэшированные данные и освобождает место в хранилище в фоновом режиме.
    • Уничтожает df: df и любые ранее полученные из него DataFrames больше не пригодны для использования, и любая последующая операция над ними вызывает CACHE_DATAFRAME_ACCESS_AFTER_UNCACHE. Если кэшировать DataFrame, полученный из df до вызова df.unpersist(), этот DataFrame остаётся доступным, потому что у него есть собственный снимок.
    • Чтобы продолжать работать с тем же источником данных, создайте новый DataFrame из исходного источника, вместо повторного использования некэшированного DataFrame.
    • Вызов df.unpersist() для DataFrame, который никогда не кэшировался или кэш которого уже был очищен, не выполняет никаких действий.

Вызовите 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.

  • Возвращает: 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

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. Тебе не нужны дополнительные привилегии.

Кэшированные данные обеспечивают изоляцию на уровне сессии. Только сессия, которая кэшировала DataFrame, может читать его кэшированные данные. Другие пользователи не могут получить к нему доступ, включая других пользователей того же блокнота.

Кэшированные данные содержат то, что вам было разрешено прочитать при кэшировании. Изменения разрешений, фильтры строк и маски столбцов, впоследствии применённые к исходным таблицам, не применяются к кэшированным данным.

Лучшие практики

  • Используйте кэш, когда нужен стабильный снимок: если вы запускаете несколько преобразований, агрегаций или детализаций одних и тех же вычисленных данных, снимок даёт им стабильную базовую линию, которая не меняется при изменении исходных таблиц.
  • Кэшировать дорогие вычисления, а не простые чтения таблиц: df.cache() материализует результат запроса, а последующие чтения сканируют этот снимок. Если вы кэшируете простое чтение таблицы, например spark.table("main.sales.orders").cache(), поздние чтения всё равно сканируют полную копию таблицы, поэтому кэширование даёт мало пользы. Кэширование оправдывает себя после выполнения ресурсоёмких операций, таких как соединения, агрегации и сложные фильтры.
  • Не кэшируйте данные, которые вы прочитали только один раз: df.cache() сразу вычисляет и сохраняет результат, что требует времени и памяти. Если прочитать результат всего один раз, первоначальная стоимость того не стоит.
  • Используйте кэш, работая на основе кэшированного DataFrame: чтобы прочитать снимок, постройте последующие операции на основе кэшированной переменной DataFrame, например df.select(...) или df.filter(...). Независимый DataFrame, построенный на основе тех же вычислений в исходных таблицах, не использует кэш.
  • Не используйте кэш для обмена данными между сеансами или между пользователями: кэшированные данные доступны только в рамках вашего сеанса. Другие сессии, блокноты, запуски заданий и пользователи не могут прочитать его. Чтобы делиться вычислёнными результатами между сессиями или с другими пользователями, записывайте их в постоянную таблицу.
  • Держите сессии с кэшированными данными менее семи дней: см. Жизненный цикл и временные ограничения. Если вашему рабочему процессу нужны данные дольше, запишите их в таблицу.
  • Вызовите df.unpersist(), когда кэшированные данные больше не нужны: это позволяет Azure Databricks раньше освободить место в хранилище. После вызова DataFrame и DataFrames, полученные из него, нельзя использовать.
  • Не используйте df.unpersist() для обновления данных: df.unpersist() завершает время жизни снимка. Чтобы учитывать изменения в исходных таблицах, постройте новый DataFrame из исходных таблиц.

Ограничения

Кэширование DataFrame на бессерверных вычислениях имеет следующие ограничения:

  • Имена и типы столбцов: следующие DataFrames не подлежат кэшированию. Переименуйте или приведите типы столбцов перед кэшированием, например, с помощью .alias("total_amount").
    • DataFrames с дублирующими названиями столбцов.
    • DataFrames с названиями столбцов, содержащими пробелы или любые символы ,;{}()=, например sum(amount), полученными при агрегации.
    • DataFrames с интервальными столбцами.
    • DataFrames с массивами, ассоциативными массивами или структурами, содержащими значения типа NULL.
    • DataFrames без столбцов или только с столбцами, типированными 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, используйте сеансовую временную таблицу.
  • Запросы к информационной схеме: Когда в метамагазине много сессий кэшируют DataFrames, запросы к information_schema.volumes и information_schema.schemata могут быть медленными или завершаться по тайм-ауту. Фильтрация volumes по каталогу не ускоряет запрос. Запросы information_schema.tables не затрагиваются.
  • Azure для государственных организаций regions: кэширование DataFrame не поддерживается на бессерверных вычислениях в регионах Azure для государственных организаций.
  • Игнорируемые параметры: df.persist(storageLevel) игнорирует уровень хранения, и df.unpersist(blocking) игнорирует blocking. См. df.persist(storageLevel) и df.unpersist(blocking).

Переход с классических вычислений

Для большинства рабочих нагрузок кэширование на бессерверных вычислениях даёт те же результаты, что и в классических вычислениях, и существующий код выполняется без изменений. В следующей таблице перечислены различия.

Behavior Бессерверные вычисления Классические вычисления
Совпадение кэша Только кэшированный DataFrame и DataFrames, полученные из него, автоматически используют кэш. Любой запрос с одинаковой формой плана автоматически использует кэш.
Свежесть данных Согласованный снимок, сделанный в момент вызова df.cache(). Варьируется: кэшированные данные обновляются в некоторых случаях, а в других случаях отображается снимок.
Область совместного доступа Только для вашего сеанса. Разделяется между сессиями внутри кластера.
Материализация Eager: вычисляется сразу, когда вы вызываете df.cache(). Ленивые: вычисляется при первом же действии на DataFrame.
df.unpersist() Уничтожает DataFrame и детерминированно освобождает связанный снимок. Освобождает кэш в оперативной памяти. При последующем использовании DataFrame могут быть получены самые свежие или устаревшие данные, в зависимости от продукта и источника данных.

Ниже приведены первые два различия более подробно:

  • Сопоставление кэша:
    • Нагрузки, затронутые изменением: большинство рабочих нагрузок не затронуты. Если вы продолжаете использовать кэшированную переменную DataFrame, например с df.select(...) или df.filter(...), кэш работает как раньше. Это изменение касается только тех рабочих нагрузок, где построенный отдельно DataFrame или SQL-запрос повторно использует кэш из-за той же формы плана. При бессерверных вычислениях эти запросы читают исходные таблицы.
    • Если затронут ваш рабочий процесс: используйте кэшированную переменную DataFrame там, где нужен кэшированный результат, вместо того чтобы снова создавать тот же запрос.
  • Свежесть данных:
    • Рабочие нагрузки, затронутые изменением: большинство рабочих нагрузок не затронуто, поскольку повторное чтение кэшированных данных после записи в исходные таблицы встречается редко. Изменение касается только рабочих нагрузок, которые записывают в исходные таблицы и затем ожидают, что кэшированный DataFrame покажет новые данные.
    • Если это влияет на вашу рабочую нагрузку: после записи в исходные таблицы создайте новый DataFrame из них вместо повторного использования кэшированного.

Дополнительные ресурсы

  • Последнее обновление: