Блок 2 - Оптимизация и производительность

Что такое Data Skew и как с ним бороться?¶

Data Skew (перекос данных) - ситуация, когда одна или несколько партиций содержат несоразмерно больше данных, чем остальные. В Spark UI видно: большинство Tasks завершились за 2 секунды, одна - за 10 минут.

Способы борьбы:

Разница между cache() и persist()¶

cache() - синоним persist(StorageLevel.MEMORY_AND_DISK). persist() позволяет указать уровень хранения явно.

Когда использовать: DataFrame используется в нескольких Action-ах или итерируется в цикле. После использования - df.unpersist(), чтобы освободить память.

Нетривиальный gotcha с input_file_name(): функция F.input_file_name() возвращает пустую строку, если вызвать её после cache(), потому что кеш разрывает связь с исходным файлом в плане выполнения. Колонку с именем файла нужно добавить до кеширования:

# ❌ Неверно - input_file_name() вернёт пустую строку
df = spark.read.parquet("s3a://bucket/data/")
df.cache()
df = df.withColumn("source_file", F.input_file_name())  # всегда ""

# ✓ Верно - добавить до cache()
df = spark.read.parquet("s3a://bucket/data/") \
    .withColumn("source_file", F.input_file_name())
df.cache()

Тот же эффект даёт вызов .limit() перед input_file_name() - он также разрывает lineage к источнику.

Что такое Adaptive Query Execution (AQE)?¶

AQE - оптимизатор, который перепланирует запрос в рантайме на основе реальной статистики партиций после каждого shuffle. Включён по умолчанию с Spark 3.2 (spark.sql.adaptive.enabled=true).

Три ключевых возможности:

1. Adaptive partition coalescing - уменьшает число shuffle partitions после сбора реальных данных (избегает миллионов мелких партиций)
2. Adaptive join switching - меняет Sort-Merge Join на Broadcast Join если одна сторона оказалась маленькой
3. Skew Join - автоматически разбивает hot partitions на подзадачи

Как работает Broadcast Join и когда его использовать?¶

При Broadcast Join маленькая таблица целиком копируется в память каждого Executor'а - shuffle не нужен. Spark сравнивает каждую строку большой таблицы с локальной копией маленькой.

from pyspark.sql.functions import broadcast
result = big.join(broadcast(small), "id")

Порог: spark.sql.autoBroadcastJoinThreshold (по умолчанию 10 МБ). При AQE Spark может повышать этот порог автоматически. Broadcast не работает для full outer join и при очень большом числе Executor'ов (сеть под нагрузкой).

Разница между repartition() и coalesce()¶

coalesce(1) перед записью - быстрый способ получить один файл, но вся нагрузка ляжет на один Executor.

Что такое Predicate Pushdown и Column Pruning?¶

Обе оптимизации Catalyst направлены на уменьшение объёма данных, читаемых с диска.

Predicate Pushdown - фильтры (WHERE, filter()) применяются как можно раньше, на уровне источника данных. Для Parquet: Spark передаёт фильтры в reader, который пропускает Row Groups, не удовлетворяющие условию (не читает их с диска). Для JDBC: фильтр преобразуется в SQL WHERE и выполняется в БД.

Column Pruning - Spark читает только те колонки, которые реально нужны для запроса, остальные пропускаются на этапе чтения. Parquet - колоночный формат, поэтому неиспользуемые колонки физически не читаются с диска.

# Оба механизма работают автоматически
# Плохо: читает все колонки, потом фильтрует в Spark
df = spark.read.parquet("events/").filter("date = '2024-01'")

# Хорошо: тот же код, но Catalyst автоматически:
# 1. Pushdown: читает только row groups с date = '2024-01'
# 2. Pruning: если дальше идёт .select("user_id", "amount"), читает только эти колонки
df.filter("date = '2024-01'").select("user_id", "amount")

Проверить в explain(): строки PushedFilters и ReadSchema покажут, что реально попало в reader.

df.filter("date = '2024-01'").select("user_id").explain()
# Parquet: ReadSchema: struct<user_id:string>  ← только нужная колонка
# PushedFilters: [IsNotNull(date), EqualTo(date,2024-01)]

Как использовать Spark UI для поиска узких мест?¶

1. Jobs → найти долгий Job → перейти в его Stages
2. Stages → колонка Duration → найти Stage с максимальным временем
3. Tasks → смотреть на Task Duration распределение - если одна Task сильно длиннее → skew; смотреть на Shuffle Read Size - если неравномерно → skew на читателе
4. SQL → физический план → найти Exchange (shuffle), BroadcastHashJoin, SortMergeJoin
5. Executors → колонка GC Time > 10% от Task Time → проблемы с памятью

Что такое Salting и как оно помогает?¶

Salting - добавление случайного числа к ключу Join для равномерного распределения hot-key строк по партициям.

from pyspark.sql.functions import concat, col, lit, floor, rand, explode, array

SALT_FACTOR = 10

# Большая таблица: добавляем случайный суффикс
big = big.withColumn("salted_key",
    concat(col("key"), lit("_"), (rand() * SALT_FACTOR).cast("int").cast("string")))

# Маленькая таблица: дублируем строки для каждого значения соли
small = small.withColumn("salt", explode(array([lit(i) for i in range(SALT_FACTOR)])))
small = small.withColumn("salted_key",
    concat(col("key"), lit("_"), col("salt").cast("string")))

result = big.join(small, "salted_key").drop("salted_key", "salt")

Параметр spark.sql.shuffle.partitions: на что влияет?¶

Определяет число партиций после каждого Shuffle (groupBy, join, distinct). По умолчанию: 200. Слишком мало → большие партиции, риск OOM. Слишком много → overhead на планирование, мелкие файлы при записи. С AQE (spark.sql.adaptive.enabled=true) Spark динамически уменьшает число партиций после оценки реального объёма. Хороший starting point: целевой_объём_данных_GB * 10 (100 ГБ → 1000 партиций).

Как избежать сериализации больших объектов в замыканиях?¶

Если замыкание (closure) UDF ссылается на большой Python-объект (словарь, список), Spark сериализует его и рассылает на все воркеры при каждом запуске. Решения:

# Плохо: большой dict сериализуется в каждую Task
lookup = {1: "a", 2: "b", ...}  # 100 МБ
df.filter(lambda row: lookup.get(row.id))

# Хорошо: передать как broadcast-переменную
bc = spark.sparkContext.broadcast(lookup)
df.rdd.filter(lambda row: bc.value.get(row.id))

Для DataFrame лучше сделать lookup-таблицу DataFrame'ом и использовать broadcast() при join.

Зачем использовать Broadcast-переменные?¶

Broadcast-переменная (sc.broadcast(value)) - механизм эффективной доставки read-only данных на все Executor'ы. Данные передаются один раз через BitTorrent-подобный протокол, кешируются в памяти каждого Executor'а и переиспользуются всеми Tasks этого Executor'а. Без broadcast каждая Task получила бы копию через Shuffle-механизм.

bc_dict = spark.sparkContext.broadcast({"US": "United States", "RU": "Russia"})

@udf("string")
def expand_code(code):
    return bc_dict.value.get(code, code)

Что такое Accumulators?¶

Accumulators - распределённые счётчики/агрегаторы, которые можно безопасно обновлять из Task'ов и читать только на Driver. Гарантируют корректность при повторном выполнении Tasks (задача idempotent). Встроенные типы: LongAccumulator, DoubleAccumulator. Можно создавать кастомные.

error_count = spark.sparkContext.accumulator(0)

def process(row):
    if row.value is None:
        error_count.add(1)
    return row

df.foreach(process)
print(f"Errors: {error_count.value}")

Влияние формата хранения на скорость¶

Parquet - де-факто стандарт для аналитики. Avro - для Kafka-стриминга (schema registry). CSV - только для обмена с внешними системами.

Что такое Bucketing и чем отличается от Partitioning?¶

При Bucketing обе таблицы должны быть bucketed по одному ключу с одинаковым N. Тогда join("user_id") не создаёт Exchange в плане.

Как работает Dynamic Resource Allocation?¶

DRA (spark.dynamicAllocation.enabled=true) позволяет Spark добавлять и убирать Executor'ы в рантайме в зависимости от нагрузки. Нет Task'ов в очереди → Executor'ы освобождаются. Очередь Tasks растёт → новые Executor'ы запрашиваются у Cluster Manager. Работает с YARN и Kubernetes. Требует spark.shuffle.service.enabled=true (внешний shuffle service) или spark.dynamicAllocation.shuffleTracking.enabled=true (Spark 3.x).

Проблема Small Files и как её решить¶

Много мелких файлов → NameNode (HDFS) или Object Store (S3) перегружены метаданными, Spark тратит секунды на LIST, каждый файл = отдельная Task → планировщик перегружен.

Решения:

• При записи: coalesce(N) или repartition(N) перед write()
• После записи: compaction job (df.repartition().write.parquet(...))
• Iceberg: spark.sql("CALL system.rewrite_data_files('table')")
• AQE coalescing: автоматически уменьшает число файлов через коалесцинг shuffle-партиций

Разница между map() и mapPartitions()¶

map(f) вызывает функцию f для каждой строки. mapPartitions(f) получает итератор всей партиции - позволяет открыть соединение один раз на партицию (DB connection pool паттерн).

def write_to_db(rows):
    conn = db.connect()  # Одно соединение на всю партицию
    for row in rows:
        conn.insert(row)
    conn.close()
    return []

df.rdd.mapPartitions(write_to_db).count()

В DataFrame API эквивалент - foreachPartition().

Что такое Speculative Execution?¶

При spark.speculation=true Spark запускает дублирующую Task для медленных (straggler) Tasks - тех, что выполняются значительно дольше медианы Stage'а. Первая завершившаяся копия принимается как результат, вторая убивается. Помогает при временных аппаратных проблемах (slow disk, network jitter), но бесполезен при настоящем data skew - дубль тоже будет медленным.

Как настроить память Executor'а?¶

spark.executor.memory     = heap память (для хранения и вычислений)
spark.executor.memoryOverhead = off-heap (JVM metadata, native библиотеки)
spark.memory.fraction     = 0.6 (доля heap для execution + storage, остаток - user code)
spark.memory.storageFraction = 0.5 (из fraction - доля для cache)

Практическая формула для контейнера 8 ГБ:

• executor.memory = 6g (6 ГБ heap)
• executor.memoryOverhead = 1g (JVM + Python worker)
• executor.cores = 4 (правило: не больше 5 cores на Executor)
• Итого: 7 ГБ из 8 ГБ контейнера (запас на OS)

Как работает Cost-Based Optimizer (CBO)?¶

CBO анализирует статистику таблиц (число строк, уникальных значений, размер колонок) и использует её для выбора оптимального плана - в отличие от Rule-Based Optimizer, который применяет правила независимо от объёма данных.

CBO помогает Spark принимать лучшие решения о порядке join-ов, выборе стратегии join и размещении broadcast-таблиц.

-- Собрать статистику для CBO
ANALYZE TABLE users COMPUTE STATISTICS;
ANALYZE TABLE users COMPUTE STATISTICS FOR COLUMNS user_id, age, country;

# Включить CBO (по умолчанию включён в Spark 3.x)
spark.conf.set("spark.sql.cbo.enabled", "true")
spark.conf.set("spark.sql.statistics.histogram.enabled", "true")

# Проверить что статистика учитывается
df.explain(mode="cost")  # Покажет estimated rows и sizes

Практический эффект заметен при join-ах нескольких таблиц - CBO может перестроить порядок join'ов, начав с меньшей таблицы.

В чём опасность чрезмерного кеширования данных?¶

Кеш занимает Unified Memory Executor'а - ту же область, что используется для выполнения операций (join, sort, aggregation). При нехватке execution memory Spark вынужден делать spill на диск, что замедляет работу.

Конкретные риски:

• Execution memory вытесняется → join/aggregation начинают spill → замедление
• GC pressure → большой heap с объектами → длинные паузы GC
• Stale данные - кеш не обновляется автоматически, если источник изменился
• LRU-вытеснение → при нехватке памяти Spark вытесняет старый кеш, потом его же пересчитывает

Правила:

• Кешировать только то, что реально используется ≥2 раз в pipeline
• После использования вызывать df.unpersist(blocking=True)
• Не кешировать в начале pipeline - лучше ближе к месту повторного использования
• Проверять Spark UI → Storage - сколько памяти занято кешем

Как работает Spill на диск во время Shuffle или агрегации?¶

Когда данные для операции (sort, hash aggregation, build-side таблица join'а) не помещаются в execution memory Executor'а, Spark сбрасывает часть данных на локальный диск и продолжает работу:

1. Executor заполняет буфер в памяти до лимита
2. Сериализует буфер и записывает в spill-файл на spark.local.dir
3. После завершения - читает все spill-файлы + оставшееся в памяти, выполняет N-way merge

Как обнаружить: Spark UI → Stages → колонки Spill (Memory) и Spill (Disk). Если Spill (Disk) > 0 - есть spill.

Решения:

# Увеличить execution memory
spark.conf.set("spark.executor.memory", "8g")

# Уменьшить storage fraction (больше памяти для execution)
spark.conf.set("spark.memory.storageFraction", "0.3")  # было 0.5

# Больше партиций → меньше данных в каждой → меньше spill
spark.conf.set("spark.sql.shuffle.partitions", "500")

Как определить оптимальное количество ядер (CPU cores) для Executor'ов?¶

Практическое правило: не более 5 cores на один Executor. При большем числе cores HDFS-клиент начинает испытывать конкуренцию за соединения, а GC-паузы затрагивают больше одновременных Tasks.

Формула для выделения Executor'ов на узле:

Cores на узле: 16
Cores на Executor: 5
Executor'ов на узле: (16 - 1) / 5 = 3 (1 core оставить на OS/YARN)
Executor memory = (total_node_RAM - OS_RAM) / 3

# Типичная конфигурация для узла 16 CPU / 64 GB RAM
conf.set("spark.executor.cores",   "5")
conf.set("spark.executor.memory",  "18g")   # (64-4) / 3 ≈ 20GB, минус overhead
conf.set("spark.executor.memoryOverhead", "2g")

С Dynamic Allocation число Executor'ов подстраивается под нагрузку - фиксировать их количество нет необходимости.

Что такое Z-Ordering и как он помогает в оптимизации запросов?¶

Z-Ordering (Z-order curve, Morton code) - алгоритм многомерного упорядочивания данных, который позволяет файлам с нужными данными физически располагаться рядом на диске.

Обычная сортировка работает по одной колонке. Z-ordering объединяет несколько колонок в один ключ, сохраняя пространственную близость данных по всем измерениям сразу.

# Delta Lake
spark.sql("OPTIMIZE events ZORDER BY (user_id, event_date)")

# Iceberg
spark.sql("""
  CALL system.rewrite_data_files(
    table => 'events',
    strategy => 'sort',
    sort_order => 'zorder(user_id, event_date)'
  )
""")

Как помогает: файлы содержат более однородные данные по обеим колонкам → при фильтре WHERE user_id = 123 AND event_date = '2024-01-01' Spark пропускает больше файлов (data skipping по min/max статистикам).

Когда не помогает: для high-cardinality колонок с высоко случайным распределением (UUID) - эффект минимален. Сортировка по одной колонке эффективнее Z-ordering при фильтре только по ней.

Что такое Off-heap memory и когда её стоит настраивать?¶

Стандартная Java-куча (heap) управляется JVM и сборщиком мусора (GC). Когда объектов много - GC-паузы достигают сотен миллисекунд, что сказывается на скорости.

Off-heap memory - память вне JVM heap, выделяемая напрямую через sun.misc.Unsafe. GC её не видит и не трогает. Spark использует её в Tungsten для хранения данных в бинарном формате UnsafeRow.

Executor Memory Layout:
┌─────────────────────────────────────────────┐
│ JVM Heap (spark.executor.memory = 6g)       │
│   ├── Unified Memory (execution + storage)  │
│   └── User Memory (JVM objects, UDF state)  │
├─────────────────────────────────────────────┤
│ Off-Heap (spark.executor.memoryOverhead)    │  ← Python worker, native libs
│ Off-Heap Execution (spark.memory.offHeap.*) │  ← Tungsten, explicit off-heap
└─────────────────────────────────────────────┘

Два вида off-heap:

1. spark.executor.memoryOverhead - всегда нужен: Python worker-процессы, JVM metadata, native libraries. Правило: max(384 МБ, 10% от executor.memory).
2. spark.memory.offHeap.enabled=true + spark.memory.offHeap.size - явный off-heap для execution. Помогает при частых GC-паузах на больших heap'ах (>16 ГБ).

# Когда включать off-heap execution:
# GC Time > 10% в Spark UI → переносим часть execution memory в off-heap
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size",    "4g")
spark.conf.set("spark.executor.memory",        "8g")   # Heap уменьшить
spark.conf.set("spark.executor.memoryOverhead", "2g")  # Python + JVM meta

В большинстве случаев memoryOverhead достаточно - явный off-heap нужен только при реальных проблемах с GC на больших Executor'ах.

Как исправить java.io.NotSerializableException?¶

java.io.NotSerializableException означает, что в замыкании (closure) UDF или трансформации есть объект, который нельзя сериализовать для передачи на Executor.

Причина: Spark сериализует весь контекст функции (замыкание) и отправляет на каждый Executor. Если в замыкании есть non-serializable объект (JDBC Connection, файловый дескриптор, ML-модель без __reduce__), - ошибка.

# Неверно: объект conn создаётся на Driver, не сериализуется
class DBWriter:
    def __init__(self):
        self.conn = psycopg2.connect(...)  # non-serializable!

writer = DBWriter()
df.foreach(lambda row: writer.write(row))  # → NotSerializableException

# Верно: создаём объект на Executor внутри mapPartitions
def write_partition(rows):
    conn = psycopg2.connect(...)  # Создаётся на Executor, не сериализуется
    for row in rows:
        conn.execute("INSERT ...", row)
    conn.close()

df.rdd.mapPartitions(write_partition)
# Или: df.foreachPartition(write_partition)

Паттерн: Любой non-serializable объект - создавать внутри mapPartitions/foreachPartition прямо на Executor'е.

Что означает WARN TaskSetManager: task of very large size?¶

Предупреждение WARN TaskSetManager: task of very large size 100500 KB означает, что сериализованный Task (включая замыкание) занимает сотни КБ или МБ. Это приводит к избыточному трафику Driver → Executor'ы и замедлению планирования.

Причина: В замыкании UDF или трансформации есть ссылка на большую Python-коллекцию (словарь, список), которая сериализуется и отправляется в каждый Task отдельно.

# Проблема: большой словарь сериализуется в каждую Task
big_lookup = {str(i): f"value_{i}" for i in range(1_000_000)}  # ~100 МБ

@udf("string")
def enrich(key):
    return big_lookup.get(key)  # словарь попадает в каждый Task!

# Решение 1: Broadcast variable (отправляется один раз на Executor)
bc = spark.sparkContext.broadcast(big_lookup)

@udf("string")
def enrich(key):
    return bc.value.get(key)

# Решение 2: Создавать объект на Executor в mapPartitions
def process_partition(rows):
    lookup = load_lookup_from_db()  # Загружается один раз на партицию
    for row in rows:
        yield enrich_row(row, lookup)

df.rdd.mapPartitions(process_partition)

Как работает AQE детально - опишите 4-шаговый цикл?¶

AQE (Adaptive Query Execution) переоптимизирует план запроса в рантайме после каждого shuffle/exchange.

4-шаговый цикл для каждого Stage:

1. Выполнить текущий Stage (compute)
   ↓
2. Shuffle Write / Broadcast + сбор реальной статистики партиций
   (знаем точный размер каждой partition - не оценочный из ANALYZE TABLE)
   ↓
3. Re-оптимизация оставшегося (невыполненного) плана
   на основе РЕАЛЬНЫХ данных:
   - Если partition стала мала → Coalescing (объединить несколько в одну)
   - Если таблица после фильтра < threshold → Switch SortMerge → BroadcastHash
   - Если partition в 5× больше медианы → SkewJoin (дробить + дублировать)
   ↓
4. Выполнить следующий Stage с обновлённым планом
   ↓ (повторить для каждого Exchange узла)

Ключевое отличие от CBO: CBO работает до выполнения на основе ANALYZE TABLE (которая может быть устаревшей). AQE работает во время выполнения на основе реальных данных. Оба механизма дополняют друг друга.

spark.conf.set("spark.sql.adaptive.enabled", "true")  # default с Spark 3.2
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")