Кэширование: cache(), persist() и управление памятью

Как Spark кэширует данные через BlockManager, все Storage Levels с механикой, LRU eviction, unpersist и когда кэш вреден.

core optimization

Зачем нужно кэширование¶

Lazy Evaluation: каждый Action пересчитывает всё заново¶

Spark работает на принципе ленивых вычислений: трансформации (filter, join, withColumn) лишь строят логический план - граф зависимостей, называемый Lineage. Реальные вычисления начинаются только при вызове Action (count, show, write). Это умно: Spark объединяет трансформации в оптимизированный план и избегает лишней работы.

Но у этого есть серьёзная оборотная сторона: при каждом Action Spark заново выполняет весь Lineage от самого начала - от чтения исходного файла до последней трансформации.

# Дорогой pipeline: чтение S3 + JOIN + Python UDF
df_heavy = (
    spark.read.parquet("s3://datalake/events/2024/")   # Чтение терабайта данных
    .join(users_df, "user_id")                          # Shuffle по сети
    .withColumn("category", classify_udf("event_type")) # Медленный Python UDF
    .filter("category IS NOT NULL")
)

# БЕЗ кэша: весь pipeline выполняется дважды
count = df_heavy.count()      # Action 1 → S3 scan + JOIN + UDF → 10 минут
df_heavy.write.parquet("out/") # Action 2 → S3 scan + JOIN + UDF → ещё 10 минут
# Итого: 20 минут вместо 10

Описание проблемы: df_heavy.count() запускает полный execution plan - Spark читает данные с S3, выполняет Shuffle для JOIN, прогоняет все строки через Python UDF. Когда затем вызывается df_heavy.write, Spark не помнит результат count(). Он начинает сначала: снова S3, снова JOIN, снова UDF. Вы платите двойную цену.

Точка ветвления: главный сигнал к кэшированию¶

Диаграмма показывает разницу: слева - три Action вызывают три полных прохода через дорогой pipeline. Справа - первый Action материализует кэш в BlockManager; второй и третий Action читают данные оттуда, минуя повторный S3-scan, JOIN и UDF.

Два главных сигнала к кэшированию:

Точка ветвления: один DataFrame используется в нескольких Action или ветках вычислений.
Дорогая трансформация: pipeline включает Shuffle (JOIN, groupBy), Python UDF, медленные сканы или агрегации с большим количеством данных.

Если DataFrame используется только один раз - кэш бесполезен и вреден: он тратит время на запись и занимает память.

Архитектура: как Spark хранит кэш¶

BlockManager: распределённая файловая система в памяти¶

Кэширование в Spark реализовано через BlockManager - компонент, работающий на каждом Executor и на Driver. BlockManager управляет хранением блоков данных (партиций) в RAM и на диске, а также их репликацией между узлами.

BlockManagerMaster на Driver'е знает, где хранится каждый блок-партиция: на каком Executor, в RAM или на диске, сколько реплик. Когда задаче нужна партиция из кэша - она запрашивает у Master её location, затем читает напрямую с нужного Executor через BlockTransferService (Netty).

Storage Memory - часть кучи JVM, выделенная для хранения кэшированных партиций. Её размер: spark.executor.memory × spark.memory.fraction × (1 - spark.memory.storageFraction). При нехватке места включается механизм вытеснения (eviction).

Каждая партиция хранится как отдельный блок с именем вида rdd_N_P (номер RDD и номер партиции). Блок может быть:

в Java heap - десериализованные объекты (MEMORY_ONLY)
в off-heap - нет (для DataFrame/RDD)
в виде byte[] в heap - сериализованный (MEMORY_ONLY_SER)
на локальном диске Executor'а - spill файл

cache() vs persist()¶

Что делает cache()¶

cache() - это метод-псевдоним, который вызывает persist() с уровнем по умолчанию. В Spark 3.x для DataFrame этот уровень - MEMORY_AND_DISK.

# Эти два вызова полностью идентичны для DataFrame в Spark 3.x:
df.cache()
df.persist()                                  # тот же дефолтный уровень
df.persist(StorageLevel.MEMORY_AND_DISK)      # явно

# Проверить текущий уровень хранения
print(df.storageLevel)
# StorageLevel(disk=True, memory=True, offHeap=False, deserialized=True, replication=1)

Исторически в Spark 2.x и для RDD дефолт persist() без аргументов был MEMORY_ONLY. Для DataFrame API в Spark 3.x дефолт изменён на MEMORY_AND_DISK - это более безопасный выбор, потому что данные не теряются при нехватке памяти.

Что делает persist()¶

persist(storageLevel) позволяет явно выбрать стратегию хранения. Это важно, когда:

RAM ограничена и нужно экономить через сериализацию (_SER)
Pipeline критичен и нужна отказоустойчивость (_2)
Данные слишком большие для RAM и выгоднее хранить только на диске (DISK_ONLY)

from pyspark import StorageLevel

# Явные уровни:
df.persist(StorageLevel.MEMORY_ONLY)           # только RAM, без disk fallback
df.persist(StorageLevel.MEMORY_AND_DISK)       # RAM + overflow на диск (default)
df.persist(StorageLevel.MEMORY_ONLY_SER)       # RAM, сериализованный (compact)
df.persist(StorageLevel.MEMORY_AND_DISK_SER)   # RAM + диск, всё сериализовано
df.persist(StorageLevel.DISK_ONLY)             # только диск
df.persist(StorageLevel.MEMORY_AND_DISK_2)     # RAM + диск + 2 реплики

# Создать собственный уровень (редко нужно)
custom_level = StorageLevel(
    useDisk=True,
    useMemory=True,
    useOffHeap=False,
    deserialized=True,
    replication=2
)
df.persist(custom_level)

Ленивость: кэш не материализуется до первого Action¶

Вызов .cache() или .persist() - это трансформация. Spark просто добавляет пометку в план: «при первом вычислении этого DataFrame - сохрани результат». Реальная работа начинается только при Action.

df = spark.read.parquet("huge_dataset/")
df_filtered = df.filter("status = 'active'").join(dim, "id")

# Ничего не происходит - просто пометка в плане
df_filtered.cache()

# Здесь происходит кэширование:
# Spark читает данные, выполняет filter + join, записывает результат в BlockManager
df_filtered.count()   # ← первый Action = материализация кэша

# Теперь все последующие Actions читают из кэша
df_filtered.show()           # из кэша - быстро
df_filtered.write.save("…")  # из кэша - быстро

# ВАЖНО: возвращаемое значение cache()/persist() - тот же DataFrame
# Правильный паттерн:
df_cached = df_filtered.cache()  # сохраняем ссылку на кэшированный DF
df_cached.count()                # прогреваем

Storage Levels: куда и как кладём данные¶

Полная карта уровней хранения¶

MEMORY_ONLY¶

Партиции хранятся в Java heap как десериализованные Java-объекты (для DataFrame - UnsafeRow). Доступ к данным максимально быстрый - не нужна десериализация перед чтением. Но:

Если партиция не помещается в Storage Memory - она не кэшируется вообще. При следующем Access Spark пересчитает её заново через Lineage.
Десериализованные объекты занимают 2–5 раз больше места, чем их сжатая форма.
Высокое давление на GC: много объектов в heap → частые GC паузы.

df.persist(StorageLevel.MEMORY_ONLY)

# Пример: DataFrame из 8 партиций, Storage Memory = 4 ГБ, DF = 6 ГБ
# Результат: 4 ГБ помещаются → кэшируются (≈5 партиций)
#            2 ГБ не помещаются → партиции 6,7,8 пересчитываются при доступе

Когда использовать: маленькие DataFrame (< 50% Storage Memory), итеративные алгоритмы где нужна максимальная скорость чтения, данные равномерно распределены по партициям.

MEMORY_AND_DISK¶

Дефолтный уровень cache(). Работает как MEMORY_ONLY, но при нехватке RAM партиции вытесняются на локальный диск Executor'а вместо пересчёта. Это ключевое отличие: данные не теряются, пересчёта нет - просто чтение с диска медленнее.

df.cache()  # = MEMORY_AND_DISK

# Пример: DF = 10 ГБ, Storage Memory = 6 ГБ
# Результат: 6 ГБ в RAM (быстро), 4 ГБ на локальном SSD Executor'а (медленнее)
# Нет пересчёта → надёжно для production ETL

Когда использовать: production ETL/ELT пайплайны, когда надёжность важнее скорости, при неизвестном распределении данных и риске spill.

MEMORY_ONLY_SER и MEMORY_AND_DISK_SER¶

Serialized уровни хранят данные в виде byte[] - сериализованного байтового массива. Spark использует Java Serialization по умолчанию или Kryo (если настроен spark.serializer = org.apache.spark.serializer.KryoSerializer).

# Включить Kryo для более компактной сериализации
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

# Теперь SER-уровни используют Kryo (в 2–5 раз компактнее Java Serialization)
df.persist(StorageLevel.MEMORY_ONLY_SER)       # Kryo → byte[] в RAM
df.persist(StorageLevel.MEMORY_AND_DISK_SER)   # Kryo → byte[] в RAM + Disk

Trade-off: экономия памяти (в 2–4 раза меньше, чем десериализованный) против CPU cost на сериализацию/десериализацию при каждом чтении из кэша. Это имеет смысл, когда:

RAM ограничена и без сериализации многие партиции «вылетают» на диск
Данных много, но обработка CPU-лёгкая (чтение без сложных вычислений)
GC паузы мешают производительности (byte[] - один объект на партицию, GC видит 1 объект вместо миллиона строк)

# Практическое сравнение использования памяти:
# DataFrame с 100М строк (10 колонок, mixed types):
# MEMORY_ONLY:         ~8 GB (десериализованные UnsafeRow + overhead)
# MEMORY_ONLY_SER:     ~2 GB (Kryo byte[])
# Экономия: 4x при цене CPU на десериализацию каждого чтения

DISK_ONLY¶

Партиции хранятся только на локальном диске Executor'а, в сериализованном виде. В RAM они не загружаются совсем.

df.persist(StorageLevel.DISK_ONLY)

Казалось бы, зачем? Если данные на диске - не проще их пересчитать? Не всегда:

Источник данных медленный (объектное хранилище S3/GCS с высокой задержкой, сетевая NFS-файловая система)
Pipeline включает дорогие трансформации (много JOIN, UDF, агрегации), и Shuffle stage занимает больше времени, чем чтение с локального SSD
Локальный SSD Executor'а в 5–10 раз быстрее, чем S3 по сети

Когда использовать: редко. Оптимальный сценарий - очень дорогой pipeline (> 30 мин) с данными на медленном объектном хранилище, и DataFrame нужен 2–3 раза.

Replicated levels (_2)¶

Уровни с суффиксом _2 хранят каждую партицию на двух Executor'ах. Это защита от отказа узла: если один Executor упадёт - данные доступны на втором без пересчёта.

df.persist(StorageLevel.MEMORY_AND_DISK_2)
# Каждая партиция → BlockManager Executor 1 (primary) + BlockManager Executor 2 (replica)

Цена репликации: вдвое больше памяти, передача блоков по сети при записи в кэш. Оправдано, когда:

Пересчёт занимает часы (граф сложных агрегаций, ML features)
Кластер нестабилен (spot/preemptible instances с высокой вероятностью замены)
Потеря кэша при отказе узла приведёт к срыву дедлайна

Не оправдано, когда: данные можно пересчитать за 10 минут - быстрее пересчитать, чем гонять двойные блоки по сети.

Полная таблица Storage Levels¶

Level	RAM	Disk	Сер.	Реплики	Примерный размер	Когда
`MEMORY_ONLY`	✅	❌	нет	1	100%	Маленький DF, быстрый CPU, max скорость
`MEMORY_AND_DISK`	✅	overflow	нет	1	100%	Production ETL, безопасный выбор по умолчанию
`MEMORY_ONLY_SER`	✅	❌	Kryo	1	~25%	RAM tight, много итераций, GC проблемы
`MEMORY_AND_DISK_SER`	✅	overflow	Kryo	1	~25%	Компромисс RAM/CPU, большие DF
`DISK_ONLY`	❌	✅	Kryo	1	-	Медленный источник, нет RAM
`MEMORY_AND_DISK_2`	✅	overflow	нет	2	200%	Критичные pipeline, нестабильный кластер
`DISK_ONLY_2`	❌	✅	Kryo	2	-	Редко - диск + отказоустойчивость

Как выбрать Storage Level¶

Диаграмма - дерево решений для выбора Storage Level. Начинайте с вопроса: используется ли DataFrame более одного раза? Если нет - кэш бесполезен. Затем: помещается ли в RAM? При положительном ответе - MEMORY_ONLY (если нет GC проблем) или MEMORY_ONLY_SER. При нехватке RAM переходим к MEMORY_AND_DISK, который Spark использует по умолчанию через cache().

Три практических правила:

Правило 1: Начинай с MEMORY_AND_DISK (cache()) - это безопасный выбор в 90% случаев.
Правило 2: Если GC overhead > 15% execution time (видно в Spark UI) или OOM - переходи на MEMORY_AND_DISK_SER с Kryo.
Правило 3: Не используй _2 если DataFrame пересчитывается за < 30 минут - сетевой overhead не оправдывает себя.

Жизненный цикл кэша¶

Создание и прогрев (Warming up)¶

Шаг «прогрева» (df.count()) - критически важен. Без него кэш создаётся инкрементально: первый Action кэшируется + выполняет основную работу, последующие читают из кэша. Если первый Action тяжёлый (например, write к S3), то к моменту его завершения кэш готов. Но в циклах лучше явно прогреть кэш через дешёвый count():

# Паттерн «прогрев перед циклом»:
df_features = compute_expensive_features(raw_df).cache()
df_features.count()  # Прогрев: materialized один раз, дальше - из кэша

for model_version in range(10):
    model.train(df_features)   # читает из кэша - быстро
    metrics = evaluate(df_features, model)  # читает из кэша - быстро

LRU Eviction: как Spark вытесняет кэш¶

При нехватке Storage Memory Spark применяет политику LRU (Least Recently Used). Это работает на уровне партиций (блоков), а не целых DataFrame:

Если Storage Memory переполнена и нужно закэшировать новый блок - Spark ищет наименее недавно использованный блок.
Для MEMORY_ONLY: вытесненный блок просто удаляется - при следующем обращении пересчитывается.
For MEMORY_AND_DISK: вытесненный блок спиллируется на диск - при обращении читается с диска, но не пересчитывается.

# Пример: два DataFrame конкурируют за Storage Memory

df_a = compute_a().persist(StorageLevel.MEMORY_ONLY)
df_a.count()  # → занимает 6 ГБ Storage Memory

df_b = compute_b().persist(StorageLevel.MEMORY_ONLY)
df_b.count()  # → нужно 4 ГБ, но доступно только 2 ГБ

# LRU: партиции df_a (использованы раньше) вытесняются
# При df_a.show() → часть партиций df_a пересчитается заново (нет disk fallback!)

# Правильно - использовать MEMORY_AND_DISK:
df_a = compute_a().persist(StorageLevel.MEMORY_AND_DISK)
# Теперь вытесненные партиции df_a уходят на диск, не пересчитываются

Почему не стоит полагаться на автоматическое LRU: в длинных pipeline с 5–10 кэшированными DataFrame вы не контролируете, какие партиции вытесняются. Явный unpersist() гарантирует, что старый ненужный кэш не конкурирует за память с новым нужным.

unpersist(): явное освобождение¶

# Базовое освобождение (асинхронное - Spark планирует удаление)
df.unpersist()

# Синхронное - дожидается фактического удаления блоков
df.unpersist(blocking=True)

# Когда использовать blocking=True:
# - В unit-тестах, где важен предсказуемый порядок
# - В длинных pipeline перед выделением новой крупной переменной
# - Когда нужно освободить память ДО следующего шага (иначе OOM)

# Очистить ВСЕ кэшированные таблицы (SQL-уровень)
spark.catalog.clearCache()

# Очистить конкретную таблицу SQL-кэша
spark.catalog.uncacheTable("orders_cached")

Гигиена кода: всегда вызывайте unpersist() в конце блока, где DataFrame используется. В длинных ноутбуках забытые .cache() без .unpersist() накапливаются и «засоряют» Storage Memory кластера - в итоге или LRU вытесняет нужный кэш, или OOM.

# Паттерн с try/finally для гарантированного освобождения
df_expensive = compute_heavy().cache()
try:
    df_expensive.count()  # прогрев
    result_a = df_expensive.groupBy("key").count()
    result_b = df_expensive.filter("status = 'active'").write.save("…")
finally:
    df_expensive.unpersist()  # освобождается даже при исключении

Продвинутые техники¶

Eager Caching: принудительный прогрев¶

В некоторых сценариях важно, чтобы кэш был готов до начала основных вычислений:

# Итеративный алгоритм с прогревом:
def train_iterative(df_raw, n_iterations):
    # Готовим features один раз - дорого
    df_features = (
        df_raw
        .join(dim_table, "user_id")
        .withColumn("feature_1", udf_1("col_a"))
        .withColumn("feature_2", udf_2("col_b"))
        .cache()
    )

    # Eager warming - прогреваем кэш ДО цикла
    row_count = df_features.count()
    print(f"Cached {row_count} rows, starting training...")

    for i in range(n_iterations):
        # Всё чтение - из кэша
        metrics = model_step(df_features, i)
        print(f"Iteration {i}: {metrics}")

    df_features.unpersist()
    return model

SQL Cache: кэширование через каталог¶

Spark позволяет кэшировать именованные таблицы и представления через SQL или Catalog API. Это полезно при работе со Spark SQL или в Jupyter-ноутбуках с BI-запросами:

# Зарегистрировать DataFrame как временную таблицу и закэшировать
df_orders = spark.read.parquet("orders/").filter("year = 2024")
df_orders.createOrReplaceTempView("orders_2024")

# Кэшировать таблицу (LAZY - как обычный cache())
spark.sql("CACHE TABLE orders_2024")

# Кэшировать немедленно (EAGER - материализует сразу без Action)
spark.sql("CACHE LAZY TABLE orders_2024")   # = lazy (DEFAULT)
spark.sql("CACHE TABLE orders_2024")        # тоже lazy в большинстве версий

# Через Catalog API:
spark.catalog.cacheTable("orders_2024")
spark.catalog.isCached("orders_2024")  # → True/False

# Освободить
spark.sql("UNCACHE TABLE orders_2024")
spark.catalog.uncacheTable("orders_2024")
spark.catalog.clearCache()  # все таблицы

Кэшированные таблицы видны в Spark UI → Storage как In-memory table orders_2024.

Cache vs Broadcast: разные задачи¶

Часто путают: когда использовать cache(), а когда broadcast()? Это разные инструменты для разных задач:

	`cache()` / `persist()`	`broadcast()`
Цель	Повторное использование одного DF несколько раз	Ускорить JOIN с маленькой таблицей
Где хранится	BlockManager каждого Executor	BlockManager (через Driver)
Размер	Любой (с disk fallback)	< 50–100 МБ
Применение	Итерации, несколько Action	JOIN с dimension-таблицей
Устраняет Shuffle	Нет	Да (для JOIN)

# cache() - для повторного использования в нескольких Action/ветках
df_features.cache()
accuracy = df_features.filter("split = 'test'").agg(F.mean("correct")).collect()
df_features.write.parquet("features_snapshot/")

# broadcast() - для JOIN без Shuffle (не имеет смысла для повторных Action)
from pyspark.sql.functions import broadcast
df_large.join(broadcast(df_small_dim), "key")

# НЕПРАВИЛЬНО: cache + broadcast - избыточно
df_small_dim.cache()
df_large.join(broadcast(df_small_dim), "key")
# broadcast() уже копирует данные на все Executor'ы - cache() здесь не нужен

Explain plan и InMemoryTableScan¶

Когда DataFrame закэширован, физический план содержит специальный узел InMemoryTableScan вместо обычного FileScan:

df_cached = df.cache()
df_cached.count()  # материализуем

df_cached.filter("region = 'EU'").explain("formatted")

== Physical Plan ==
*(1) Filter (region#0 = EU)
+- *(1) InMemoryTableScan [id#1, region#0, amount#2], [isnotnull(region#0), (region#0 = EU)]
      +- InMemoryRelation [id#1, region#0, amount#2], StorageLevel(disk, memory, deserialized, 1 replicas)
            +- *(1) FileScan parquet [id#1, region#0, amount#2] ...

InMemoryTableScan - узел чтения из BlockManager. InMemoryRelation описывает сам кэш с его StorageLevel. Если этого узла нет - кэш не используется (возможно, не прогрет или evicted).

Spark UI: вкладка Storage¶

Вкладка Storage в Spark UI показывает все активные кэши:

RDD Name                    Storage Level       Cached Partitions  Fraction Cached  Size in Memory  Size on Disk
[df_features] (MEMORY_AND_DISK)  Disk Memory Deserialized 1x Replicated  200/200  100%  4.2 GB  0 B
[df_raw] (MEMORY_AND_DISK)       Disk Memory Deserialized 1x Replicated   150/200   75%  3.1 GB  850 MB

Что анализировать:

Fraction Cached < 100%: часть партиций вытеснена. Для MEMORY_ONLY - они будут пересчитаны, для MEMORY_AND_DISK - прочитаны с диска.
Size on Disk > 0: вытеснение происходит, Storage Memory недостаточно. Рассмотрите _SER уровень или уменьшение spark.sql.shuffle.partitions.
Несколько строк - несколько кэшированных DataFrame. Следите, чтобы старые не занимали место.

# Программный доступ к статусу кэша
print(df.storageLevel)
# StorageLevel(disk=True, memory=True, offHeap=False, deserialized=True, replication=1)
# → MEMORY_AND_DISK

Cache и AQE¶

Кэшированные DataFrame замораживают план: AQE не может переоптимизировать уже материализованный кэш. Это важно понимать:

# AQE активен
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Без cache: AQE адаптирует число shuffle партиций, может переключить SMJ → BHJ
df_result = df_a.join(df_b, "key").groupBy("region").count()

# С cache: план фиксируется при первой материализации
# AQE не может изменить число партиций в уже кэшированном DF
df_result_cached = df_a.join(df_b, "key").cache()
df_result_cached.count()  # план зафиксирован

# После этого: AQE не применяется к операциям НАД кэшем, но операции ПОСЛЕ - адаптируются
df_result_cached.groupBy("region").count()  # AQE работает для группировки после кэша

Кэширование промежуточных слоёв в Lakehouse ETL¶

В production ETL пайплайнах кэш полезен на стыке дорогих стадий:

# Bronze → Silver: дорогая очистка и обогащение
df_silver = (
    spark.read.format("delta").load("s3://lake/bronze/events/")
    .filter("event_date >= current_date() - 1")    # только вчерашние события
    .join(broadcast(df_device_types), "device_id")  # BHJ для маленьких справочников
    .withColumn("user_segment", segment_udf("user_features"))  # Python UDF - дорого
    .withColumn("event_hour", F.hour("event_ts"))
    .dropDuplicates(["event_id"])
)

# Кэшируем Silver ПОСЛЕ всех дорогих трансформаций
df_silver.cache()
df_silver.count()  # Прогрев - вся дорогая обработка выполняется один раз

# Несколько выходных таблиц из одного Silver - каждая читает из кэша
df_silver.filter("channel = 'mobile'") \
    .groupBy("user_segment", "event_hour") \
    .agg(F.count("*").alias("events")) \
    .write.format("delta").mode("overwrite").save("s3://lake/gold/mobile_hourly/")

df_silver.filter("channel = 'web'") \
    .groupBy("country", "event_hour") \
    .agg(F.sum("revenue").alias("total_revenue")) \
    .write.format("delta").mode("overwrite").save("s3://lake/gold/web_revenue/")

df_silver.write.format("delta").mode("overwrite").save("s3://lake/silver/events_clean/")

# Освобождаем кэш - задача выполнена
df_silver.unpersist()

Объяснение: без кэша тройная запись (mobile_hourly, web_revenue, silver) = тройной прогон через UDF, JOIN и фильтрацию. С кэшем - один прогон, три чтения из BlockManager.

Антипаттерны кэширования¶

1. Кэширование одноразового DataFrame¶

# ПЛОХО: один Action → кэш создан и сразу не нужен
df.cache()
df.write.parquet("out/")  # единственный Action
# Кэш занимает память, но никогда не используется повторно

# ХОРОШО: без кэша, если один Action
df.write.parquet("out/")

2. Кэширование огромных сырых данных¶

# ПЛОХО: кэшируем террабайт, а нужен 1%
df_raw = spark.read.parquet("huge_table_5TB/").cache()
df_raw.count()  # → 5 ТБ в RAM/Disk = всё занято

df_small = df_raw.filter("date = '2024-01-15'")  # только нужная дата
df_small.count()
df_small.write.save("…")

# ХОРОШО: кэшировать ПОСЛЕ фильтрации
df_cached = spark.read.parquet("huge_table_5TB/") \
    .filter("date = '2024-01-15'") \
    .cache()
df_cached.count()  # → кэшируем только нужные данные (0.1 ТБ)

3. Забытые кэши в ноутбуках¶

# Типичная сессия в Jupyter:
df_users = spark.read.parquet("users/").cache()
df_users.show()

# ... 50 ячеек позже ...

df_orders = spark.read.parquet("orders/").cache()
df_orders.show()

# df_users занимает Storage Memory, но давно не нужен
# Правило: всегда в последней ячейке блока:
df_users.unpersist()

4. Кэш вместо правильной архитектуры¶

# ПЛОХО: кэш как костыль для плохого pipeline
df = (spark.read.parquet("raw/")
      .join(a, "k").join(b, "k").join(c, "k").join(d, "k")  # 4 shuffle
      .cache())  # ← пытаемся залатать медленный JOIN

# ХОРОШО: сначала оптимизировать pipeline
df = (spark.read.parquet("raw/")
      .join(broadcast(a), "k")   # BHJ вместо SMJ
      .join(broadcast(b), "k")
      .join(c, "k")              # c - большая, bucketized → no shuffle
      .join(d, "k")
      .cache())  # теперь кэш реально ускоряет повторное использование

5. Кэш в маленьких кластерах¶

# Кластер: 4 Executor × 4 ГБ = 16 ГБ total
# Storage Memory (0.6 × 0.5 = 30% от heap): ~4.8 ГБ

# Кэш 8 ГБ DataFrame занимает всё Storage Memory
# Execution Memory сокращается → spill → производительность падает
# В итоге работает медленнее, чем без кэша

# Правило: не кэшировать если DF > 50% всего Storage Memory кластера

Задания для практики¶

Задание 1: Измерение выигрыша от кэширования¶

Напишите код, который сравнивает время выполнения тяжёлого pipeline с кэшем и без, выполнив три Action на одном DataFrame. Замерьте время каждого варианта.

Решение

import time
from pyspark.sql import functions as F

# Генерируем тестовые данные
df_events = spark.range(5_000_000).selectExpr(
    "id as event_id",
    "cast(id % 10000 as int) as user_id",
    "cast(id % 50 as int) as product_id",
    "rand() * 1000 as amount",
    "concat('type_', cast(id % 20 as string)) as event_type"
)

df_users = spark.range(10_000).selectExpr(
    "id as user_id",
    "concat('User_', id) as user_name",
    "cast(id % 10 as int) as segment_id"
)

# Дорогой pipeline: JOIN + оконная функция + агрегация
def build_pipeline():
    return (
        df_events
        .join(df_users, "user_id")
        .withColumn("amount_rank",
            F.rank().over(
                F.Window.partitionBy("segment_id")
                .orderBy(F.col("amount").desc())
            ))
    )

# Вариант A: БЕЗ кэша - каждый Action пересчитывает всё
print("=== БЕЗ КЭША ===")
t_start = time.time()
df_no_cache = build_pipeline()
count = df_no_cache.count()
t1 = time.time() - t_start
print(f"Action 1 (count): {t1:.1f}s → {count} rows")

t_start = time.time()
top10 = df_no_cache.filter("amount_rank <= 10").count()
t2 = time.time() - t_start
print(f"Action 2 (filter top10): {t2:.1f}s → {top10} rows")

t_start = time.time()
by_segment = df_no_cache.groupBy("segment_id").agg(F.mean("amount")).count()
t3 = time.time() - t_start
print(f"Action 3 (groupBy): {t3:.1f}s")
print(f"Суммарно без кэша: {t1 + t2 + t3:.1f}s")

# Вариант B: С КЭШЕМ - один пересчёт, два чтения из кэша
print("\n=== С КЭШЕМ ===")
df_cached = build_pipeline().cache()

t_start = time.time()
count_c = df_cached.count()  # материализует кэш + Action 1
t1c = time.time() - t_start
print(f"Action 1 (count + кэширование): {t1c:.1f}s → {count_c} rows")

t_start = time.time()
top10_c = df_cached.filter("amount_rank <= 10").count()
t2c = time.time() - t_start
print(f"Action 2 (filter top10, из кэша): {t2c:.1f}s")

t_start = time.time()
df_cached.groupBy("segment_id").agg(F.mean("amount")).count()
t3c = time.time() - t_start
print(f"Action 3 (groupBy, из кэша): {t3c:.1f}s")
print(f"Суммарно с кэшем: {t1c + t2c + t3c:.1f}s")

# Ожидаемый результат:
# Без кэша: ~15s + ~15s + ~15s = ~45s
# С кэшем:  ~15s (кэширование) + ~1s + ~1s = ~17s
# Ускорение: ~2.6x (растёт с числом Action)

df_cached.unpersist()

# Проверяем через explain - видим InMemoryTableScan
df_cached2 = build_pipeline().cache()
df_cached2.count()
df_cached2.filter("amount_rank <= 10").explain("formatted")
# Ищем: InMemoryTableScan - значит читает из кэша, не пересчитывает
df_cached2.unpersist()

Задание 2: Выбор Storage Level под ограниченную память¶

У вас есть кластер с 10 Executor'ами по 8 ГБ. DataFrame занимает 12 ГБ (больше Storage Memory кластера). Напишите код, который:

Пробует MEMORY_ONLY и показывает, что часть партиций не закэширована
Переключается на MEMORY_AND_DISK_SER с Kryo и показывает экономию памяти
Проверяет статус кэша через storageLevel

Решение

from pyspark import StorageLevel

# Настройка: Kryo для эффективной сериализации
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

# Генерируем 12 ГБ данных (примерно)
df_large = spark.range(50_000_000).selectExpr(
    "id",
    "rand() as val_a",
    "rand() as val_b",
    "concat('string_', cast(id as string)) as label",
    "cast(id % 1000 as int) as category"
)

# ШАГ 1: Пробуем MEMORY_ONLY
print("=== MEMORY_ONLY ===")
df_mo = df_large.persist(StorageLevel.MEMORY_ONLY)
df_mo.count()  # материализуем

# Смотрим storageLevel
print(f"Storage Level: {df_mo.storageLevel}")
# StorageLevel(disk=False, memory=True, offHeap=False, deserialized=True, replication=1)

# В Spark UI → Storage увидим: Fraction Cached < 100%
# Например: 120/200 partitions cached (60%)
# Остальные 40% будут пересчитываться при доступе!

df_mo.unpersist()
print("MEMORY_ONLY unpersisted")

# ШАГ 2: Переключаемся на MEMORY_AND_DISK_SER (с Kryo)
print("\n=== MEMORY_AND_DISK_SER (Kryo) ===")
df_mads = df_large.persist(StorageLevel.MEMORY_AND_DISK_SER)
df_mads.count()  # материализуем

print(f"Storage Level: {df_mads.storageLevel}")
# StorageLevel(disk=True, memory=True, offHeap=False, deserialized=False, replication=1)

# В Spark UI → Storage:
# Fraction Cached: 200/200 partitions (100%) - всё закэшировано!
# Size in Memory: ~3 GB (Kryo сжал 12 GB в ~3 GB)
# Size on Disk: остаток если RAM не хватило

# Проверяем что все партиции кэшированы: выполняем несколько Action
import time

t1 = time.time()
count1 = df_mads.filter("category = 42").count()
t1 = time.time() - t1

t2 = time.time()
count2 = df_mads.groupBy("category").count().count()
t2 = time.time() - t2

print(f"Action 1 (из кэша): {t1:.2f}s, count={count1}")
print(f"Action 2 (из кэша): {t2:.2f}s")
# Оба Action быстрые - читают из кэша, а не пересчитывают

df_mads.unpersist(blocking=True)
print("MEMORY_AND_DISK_SER unpersisted (blocking)")

# ШАГ 3: Демонстрация blocking=True vs False
df_test = df_large.cache()
df_test.count()

print("\nNon-blocking unpersist:")
df_test.unpersist(blocking=False)
# Возвращает управление немедленно, Spark удаляет блоки в фоне
print(f"Storage level after non-blocking: {df_test.storageLevel}")
# Может всё ещё показывать MEMORY_AND_DISK пока фоновое удаление не завершено

Задание 3: Оптимизация итеративного ML pipeline¶

Реализуйте итеративный pipeline для 5 «эпох» обучения (симуляция), где features DataFrame вычисляется один раз и используется 5 раз. Добавьте правильный unpersist в конце. Сравните время с и без кэша.

Решение

import time
from pyspark.sql import functions as F

# Симуляция дорогого feature engineering
def compute_features(raw_df, lookup_df):
    """Дорогой pipeline: JOIN + оконные функции + UDF"""
    from pyspark.sql import Window

    segment_window = Window.partitionBy("segment").orderBy(F.col("score").desc())

    return (
        raw_df
        .join(lookup_df, "user_id")
        .withColumn("score", F.col("amount") * F.log1p(F.col("interaction_count")))
        .withColumn("segment_rank", F.rank().over(segment_window))
        .withColumn("score_normalized",
            F.col("score") / F.max("score").over(Window.partitionBy("segment")))
        .filter("segment_rank <= 1000")  # топ-1000 в каждом сегменте
    )

# Тестовые данные
df_raw = spark.range(2_000_000).selectExpr(
    "id as event_id",
    "cast(id % 5000 as int) as user_id",
    "rand() * 500 as amount",
    "cast(rand() * 100 as int) as interaction_count"
)
df_lookup = spark.range(5_000).selectExpr(
    "id as user_id",
    "concat('seg_', cast(id % 10 as string)) as segment"
)

N_EPOCHS = 5

# ВАРИАНТ A: Без кэша (плохо)
print("=== БЕЗ КЭША ===")
t_total_no_cache = time.time()
for epoch in range(N_EPOCHS):
    t = time.time()
    df_features = compute_features(df_raw, df_lookup)
    metrics = df_features.agg(
        F.mean("score_normalized").alias("mean_score"),
        F.count("*").alias("sample_count")
    ).collect()[0]
    print(f"Epoch {epoch}: mean={metrics.mean_score:.4f}, "
          f"samples={metrics.sample_count}, time={time.time()-t:.1f}s")
print(f"Total без кэша: {time.time()-t_total_no_cache:.1f}s\n")

# ВАРИАНТ B: С кэшем (правильно)
print("=== С КЭШЕМ ===")
t_total_cache = time.time()

# Вычисляем и кэшируем один раз
df_features = compute_features(df_raw, df_lookup).cache()
t_warm = time.time()
sample_count = df_features.count()  # ПРОГРЕВ
print(f"Cache warm-up: {time.time()-t_warm:.1f}s, {sample_count} rows cached")

for epoch in range(N_EPOCHS):
    t = time.time()
    # Все эпохи читают из кэша
    metrics = df_features.agg(
        F.mean("score_normalized").alias("mean_score"),
        F.count("*").alias("sample_count")
    ).collect()[0]
    print(f"Epoch {epoch}: mean={metrics.mean_score:.4f}, "
          f"samples={metrics.sample_count}, time={time.time()-t:.1f}s")

# ОБЯЗАТЕЛЬНО: освобождаем кэш после цикла
df_features.unpersist()
print(f"Total с кэшем: {time.time()-t_total_cache:.1f}s")

# Ожидаемый результат:
# Без кэша: каждая эпоха ~20s → 5 × 20s = 100s
# С кэшем: прогрев ~20s + каждая эпоха ~1s → 20s + 5s = 25s
# Ускорение: 4x при 5 итерациях, растёт линейно с числом итераций

# Проверяем что кэш освобождён:
print(f"Storage level после unpersist: {df_features.storageLevel}")
# → StorageLevel(False, False, False, False, 1) = NONE

Задание 4: Диагностика и устранение неправильного кэширования¶

Найдите все ошибки в коде ниже и исправьте их. Код должен: кэшировать только там, где это даёт выигрыш; освобождать кэш; выбирать правильный Storage Level.

# Исходный "сломанный" код - найдите все проблемы
raw = spark.read.parquet("s3://bucket/events_5TB/").cache()
raw.count()

daily = raw.filter("date = '2024-01-15'")
daily.cache()
daily.count()

result1 = daily.filter("region = 'EU'").write.parquet("eu_out/")

result2 = daily.groupBy("product").count().show()

temp = raw.filter("date = '2024-01-14'").cache()
temp.write.parquet("yesterday/")

big_agg = raw.groupBy("date", "region").agg(F.sum("revenue")).cache()
big_agg.show()

Решение

# НАЙДЕННЫЕ ПРОБЛЕМЫ:
# 1. raw.cache() - 5 ТБ сырых данных! Занимает весь кластер, никогда не нужно целиком.
# 2. daily.cache() - правильно кэшировать filtered DF, НО daily используется только
#    2 раза: write (один Action!) + show. Для show + write кэш оправдан.
# 3. result1 = daily.filter().write - write возвращает None, не DataFrame. Присваивание бессмысленно.
# 4. temp.cache() - используется только для одного write! Кэш бесполезен.
# 5. big_agg.cache() + .show() - один Action. Кэш никогда не используется повторно.
# 6. Нигде нет unpersist() - кэши накапливаются.
# 7. raw остаётся закэшированным на весь скрипт - засоряет кластер.

from pyspark.sql import functions as F
from pyspark import StorageLevel

# ИСПРАВЛЕННЫЙ КОД:

# 1. Читаем сырые данные БЕЗ кэша (5 ТБ - слишком много)
raw = spark.read.parquet("s3://bucket/events_5TB/")

# 2. Кэшируем только отфильтрованный daily (он используется 2 раза: write + groupBy)
daily = raw.filter("date = '2024-01-15'").cache()
daily.count()  # Прогрев кэша

# 3. Два Action на daily - кэш оправдан
daily.filter("region = 'EU'").write.parquet("eu_out/")   # Action 1
daily.groupBy("product").count().show()                    # Action 2

# 4. Освобождаем daily - больше не нужен
daily.unpersist()

# 5. temp используется только один раз → НЕ кэшировать
raw.filter("date = '2024-01-14'").write.parquet("yesterday/")  # без cache

# 6. big_agg используется только для show() → НЕ кэшировать
# (один Action - кэш бесполезен)
raw.groupBy("date", "region").agg(F.sum("revenue")).show()

# ДОПОЛНИТЕЛЬНО: если daily большой и GC проблемы - использовать SER:
daily_opt = raw.filter("date = '2024-01-15'") \
    .persist(StorageLevel.MEMORY_AND_DISK_SER)  # экономим RAM через Kryo
daily_opt.count()  # прогрев

daily_opt.filter("region = 'EU'").write.parquet("eu_out/")
daily_opt.groupBy("product").count().show()

daily_opt.unpersist(blocking=True)  # synchronous для предсказуемости в скрипте

# Итог: вместо 3 ненужных кэшей (5TB + temp + big_agg) + 1 полезного без unpersist
# → 1 правильный кэш с правильным уровнем и явным unpersist