Все 5 стратегий JOIN: физика алгоритмов и управление Catalyst

Почему физическая стратегия JOIN определяет судьбу запроса¶

JOIN - самая дорогостоящая операция в распределённых системах. Неправильный выбор алгоритма превращает 30-секундный запрос в 30-минутный. Spark реализует пять физических стратегий, каждая из которых имеет принципиально разную стоимость и применима при разных условиях.

Ключевой вопрос: нужен ли Shuffle - пересылка данных по сети? Shuffle - это самая медленная операция в кластере. BHJ полностью его устраняет. SMJ и SHJ требуют Shuffle, но компенсируют предсказуемостью. BNLJ и Cartesian работают без Shuffle, но платят квадратичной сложностью.

Диаграмма показывает дерево решений Catalyst: каждый ромб - проверка условия. Если есть equi-условие и маленькая таблица вписывается в порог трансляции - Catalyst выбирает BHJ без Shuffle. При больших таблицах с обеих сторон - SMJ по умолчанию. SHJ включается только когда явно снят флаг preferSortMergeJoin. При неравенственных условиях единственный выход - BNLJ.

1. Broadcast Hash Join (BHJ)¶

Принцип: устранить Shuffle целиком¶

BHJ - самая быстрая стратегия. Вместо перемешивания большой таблицы Spark отправляет маленькую таблицу на все узлы целиком и выполняет локальный хеш-поиск. Большая таблица читается на месте, без движения по сети.

Физический алгоритм: 6 шагов¶

Шаг 1 - Collect. Driver инициирует collect() маленькой таблицы. Все партиции маленькой таблицы собираются на Driver в виде массива UnsafeRow. UnsafeRow - бинарный формат Spark: фиксированная область для примитивов + переменная область для строк. Никакого Java heap overhead - байты хранятся напрямую в long[].

Шаг 2 - Serialize. Driver сериализует UnsafeRow[] в единый byte[]. Используется Kryo или Java serialization (настраивается через spark.serializer). Итоговый byte[] - это то, что будет отправлено.

Шаг 3 - Broadcast. SparkContext.broadcast() отправляет byte[] через торрент-подобный механизм: Driver отправляет блок нескольким Executor'ам-«seed», те передают его соседям. Это предотвращает перегрузку Driver при сотнях Executor'ов. Блок кешируется на Executor'е в BlockManager.

Шаг 4 - Deserialize. Каждый Executor десериализует byte[] обратно в UnsafeRow[]. Это происходит один раз при первом обращении к broadcast-переменной, результат кешируется в ExecutorMemory.

Шаг 5 - Build. Executor строит LongToUnsafeRowMap (для числовых ключей) или UnsafeHashedRelation (для общего случая). Структура: открытая адресация (open addressing) с линейным пробированием. Ключ хешируется Murmur3, результат - индекс в массиве. Каждый slot хранит: hash-код ключа, указатель на UnsafeRow (или linked list при коллизиях). Build-фаза полностью in-memory, без спилла.

Шаг 6 - Probe. Большая таблица читается потоково (streaming): каждая строка большой таблицы поступает в LongToUnsafeRowMap.get(key) → O(1) lookup. Совпадения эмитируются как выходные строки. Большая таблица никуда не перемещается - каждый Executor обрабатывает только свою партицию.

Почему это так быстро¶

Нет Exchange (Shuffle) для большой таблицы. Нет Sort. Probe - потоковый O(1)-поиск. Итоговая сложность - O(N), где N - размер большой таблицы.

Подводные камни и ограничения¶

Память на Executor. Если broadcast-таблица 100 МБ, а Executor'ов 200 штук - это 20 ГБ суммарно. При OOM Executor'а - SparkException: Could not execute broadcast.

Порог. autoBroadcastJoinThreshold = 10 МБ по умолчанию. Spark оценивает размер таблицы по статистике (после ANALYZE TABLE или через AQE). Если статистика недоступна - размер неизвестен, broadcast не применяется.

Driver bottleneck. При collect() весь датасет проходит через Driver. Если Driver имеет ограниченную память (например, 4 ГБ) - это потолок для broadcast. Конфигурация: spark.driver.maxResultSize.

Не поддерживает full outer join без специальных условий (только left outer, right outer и inner).

CSV и форматы без статистики. Spark умеет оценивать размер таблицы по метаданным Parquet (статистика в footer) и Delta/Iceberg (transaction log). Для CSV и JSON статистики нет - Spark не знает, сколько строк вернёт файл после фильтрации. Даже если CSV содержит 5 МБ после фильтрации, Spark видит исходный размер файла (например, 2 ГБ) и отказывается от auto-broadcast. Решение - явный broadcast() hint:

# CSV: auto-broadcast НЕ сработает даже для маленьких таблиц
df_csv = spark.read.csv("reference_small.csv", header=True)
df_large.join(df_csv, "key")  # → SMJ, несмотря на маленький размер результата

# Явный hint - единственный способ принудить broadcast для CSV
df_large.join(broadcast(df_csv), "key")  # → BHJ

Риск «молчаливого» отключения broadcast при росте данных. Порог autoBroadcastJoinThreshold проверяется на момент составления плана по текущей статистике. Если справочная таблица была 8 МБ при настройке job, а через месяц выросла до 12 МБ - Spark без предупреждений переключится с BHJ на SMJ. Производительность деградирует незаметно. Для таблиц, которые могут расти, используйте явный broadcast() hint и мониторьте размер:

# Опасно: размер dim_regions растёт → в какой-то момент auto-broadcast отключится
df_facts.join(df_dim_regions, "region_id")  # молча станет SMJ

# Надёжно: явный hint + защитная проверка размера
DIM_MAX_MB = 100
dim_size_mb = df_dim_regions.count() * 8 / 1e6  # грубая оценка
if dim_size_mb > DIM_MAX_MB:
    raise ValueError(f"dim_regions ({dim_size_mb:.0f} MB) превысил broadcast-лимит!")
df_facts.join(broadcast(df_dim_regions), "region_id")

from pyspark.sql.functions import broadcast

# Автоматически (если Spark знает размер < 10 МБ)
df_orders.join(df_countries, "country_code")

# Явный hint - форсирует BHJ независимо от размера
df_orders.join(broadcast(df_countries), "country_code")

# Увеличить порог до 50 МБ для более агрессивного BHJ
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50m")

# Отключить автоматический broadcast полностью
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# SQL hint
spark.sql("""
    SELECT /*+ BROADCAST(d) */ f.order_id, d.country_name
    FROM orders f
    JOIN countries d ON f.country_code = d.code
""")

Когда BHJ - правильный выбор¶

• Таблица измерений (dimension table) в схеме «звезда»: страны, категории, статусы
• Lookup-таблицы: курсы валют, тарифы, маппинги кодов
• Таблица < 50–100 МБ при достаточной памяти Executor'ов

2. Sort Merge Join (SMJ)¶

Принцип: shuffle → sort → merge¶

SMJ - стратегия по умолчанию для production-объёмов. Её надёжность основана на трёх фазах: перемешивание данных так, чтобы строки с одинаковым ключом оказались на одном узле; сортировка каждой партиции; слияние двух отсортированных потоков.

Физический алгоритм: 3 фазы¶

Фаза 1: Exchange (Shuffle)¶

Обе таблицы проходят через HashPartitioning: каждая строка хешируется по join-ключу (Murmur3Hash), результат берётся по модулю spark.sql.shuffle.partitions (default: 200). Строки с одинаковым ключом всегда попадают в один и тот же номер партиции - и на один и тот же Executor.

Механизм Shuffle:

1. Map side (shuffle write): каждый task записывает свои строки в shuffle файлы на локальном диске, разбитые по номеру партиции-получателя. Один файл на task, внутри - блоки для каждой целевой партиции.
2. Reduce side (shuffle read): каждый task-получатель скачивает с диска нужные блоки от всех Map-задач по HTTP через BlockManager. Это и есть «сеть» - самая дорогая часть.

После Shuffle на каждом Executor'е лежат строки только «своих» ключей - из обеих таблиц.

Фаза 2: Sort¶

Каждый Executor сортирует свои партиции обеих таблиц независимо. Spark использует UnsafeExternalSorter - внешний сортировщик, работающий с UnsafeRow:

• Строки сортируются в памяти через алгоритм TimSort (гибрид MergeSort и InsertionSort). TimSort выбран, потому что реально-мировые данные часто частично упорядочены - TimSort даёт O(N) в лучшем случае.
• Сравнение ключей происходит напрямую по байтам (UnsafeRowComparator), без десериализации в Java-объекты - максимально быстро.
• Если данные не помещаются в memory budget задачи (управляется через spark.executor.memory + spark.memory.fraction) - сортировщик спиллит на диск.

Механизм Spill (сброс на диск)¶

Спилл - ключевое свойство SMJ, делающее его надёжным при любых объёмах:

1. В памяти накапливается maxRecordsInMemory строк (или до исчерпания memory budget).
2. Когда память кончается - текущий in-memory буфер сортируется и записывается в spill-файл на локальном диске в формате UnsafeRow.
3. Процесс повторяется: накопить → отсортировать → спиллить → очистить память → продолжить.
4. В конце: один или несколько spill-файлов + хвост в памяти, каждый отсортирован. Итоговая сортировка - N-way merge всех spill-файлов через min-heap (priority queue).

Конфигурация:

# Сколько памяти под сортировку (доля от executor memory)
spark.conf.set("spark.memory.fraction", "0.6")       # default 0.6
spark.conf.set("spark.memory.storageFraction", "0.5") # default 0.5

# Количество shuffle партиций (ключ для SMJ!)
spark.conf.set("spark.sql.shuffle.partitions", "400")  # default 200

Фаза 3: Merge Iterator¶

После сортировки на каждом Executor'е есть два отсортированных итератора: один для строк левой таблицы, один для правой. Merge работает по классическому алгоритму:

LEFT_cursor = first row of sorted left partition
RIGHT_cursor = first row of sorted right partition

while LEFT_cursor != EOF and RIGHT_cursor != EOF:
    if LEFT_cursor.key < RIGHT_cursor.key:
        advance LEFT_cursor
    elif LEFT_cursor.key > RIGHT_cursor.key:
        advance RIGHT_cursor
    else:  # keys match
        collect all left rows with this key → left_group[]
        collect all right rows with this key → right_group[]
        emit cross product: left_group × right_group  # nested loop for matching rows
        advance both cursors past this key

Важный момент: когда ключ повторяется в обеих таблицах (дублирующийся ключ), Spark буферизирует в памяти все строки с этим ключом из меньшей стороны (или обеих). При скошенных данных (data skew) этот буфер может вырасти до нескольких гигабайт, что вызывает OOM или спилл именно на фазе merge.

Когда использовать SMJ¶

• Обе таблицы большие (> порога broadcast)
• Production-нагрузка с неизвестным распределением данных
• Требуется надёжность: лучше медленнее, чем OOM
• Нет возможности применить bucketing

# Автоматически при больших таблицах
df_orders.join(df_customers, "customer_id")

# Явный hint
df_orders.join(df_customers.hint("MERGE"), "customer_id")

spark.sql("""
    SELECT /*+ MERGE(o, c) */ *
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
""")

# Настройка числа партиций - критична для SMJ
spark.conf.set("spark.sql.shuffle.partitions", "400")

3. Shuffle Hash Join (SHJ)¶

Принцип: shuffle без сортировки¶

SHJ - гибрид BHJ и SMJ. Данные перемешиваются как в SMJ (чтобы одинаковые ключи попали на один узел), но вместо сортировки и слияния - строится хеш-таблица по меньшей партиции, затем большая партиция пробируется через неё. Нет сортировки → быстрее SMJ при достаточной памяти.

Физический алгоритм¶

Shuffle-фаза идентична SMJ: обе таблицы хешируются по ключу, строки распределяются по партициям. Различие начинается после.

Build-фаза. Меньшая партиция (Spark выбирает автоматически) целиком загружается в UnsafeHashedRelation:

• Структура: массив фиксированного размера (open addressing), слоты = (hash, valuePointer).
• При коллизии (два ключа → один slot): linear probing - сдвиг на +1 позицию.
• Ключ хешируется дважды: Murmur3 для индекса в массиве + inline сравнение для подтверждения.
• Все строки хранятся в одном byte[] буфере, valuePointer - смещение в нём.

Probe-фаза. Большая партиция читается потоком: каждая входящая строка → hash(key) → lookup в UnsafeHashedRelation → O(1). Совпадения эмитируются немедленно без буферизации.

SHJ vs SMJ: когда выгоднее¶

SHJ выгоден, когда:

• Одна партиция после Shuffle заметно меньше другой (например, 100 MB vs 1 GB)
• Оперативной памяти достаточно для хранения hash-таблицы целой партиции
• Данные равномерно распределены (нет skew - иначе hash-таблица одной «горячей» партиции переполнит память)

SMJ надёжнее при:

• Неизвестном распределении (skew → риск OOM в SHJ)
• Очень большой меньшей стороне (hash-таблица не помещается в RAM)

Почему Spark по умолчанию предпочитает SMJ даже когда SHJ быстрее? Потому что preferSortMergeJoin = true - консервативный дефолт. SMJ гарантирует выполнение без OOM благодаря spill-механизму; SHJ может упасть при неожиданном skew.

# Включить предпочтение SHJ (использовать осторожно)
spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")

# Явный hint
df_large.join(df_medium.hint("SHUFFLE_HASH"), "key")

spark.sql("""
    SELECT /*+ SHUFFLE_HASH(s) */ *
    FROM large_table l
    JOIN medium_table s ON l.key = s.key
""")

Опасность: OOM при SHJ¶

При data skew один ключ может занять тысячи строк в hash-таблице. Если build-партиция не помещается в memory budget задачи (часть spark.executor.memory * spark.memory.fraction) - задача упадёт с OutOfMemoryError. SHJ не умеет спиллить hash-таблицу на диск (в отличие от SMJ). Диагностика: в UI смотрите на размер партиций в Shuffle Read.

4. Broadcast Nested Loop Join (BNLJ)¶

Принцип: единственный выход без equi-условия¶

BNLJ - стратегия для non-equi join: когда условие соединения содержит неравенство (>, <, BETWEEN, !=) или логически не может быть выражено через равенство ключей. Spark транслирует меньшую таблицу (как в BHJ), а затем для каждой строки большой таблицы перебирает все строки малой - вложенный цикл.

Физический алгоритм¶

Шаг 1 - Broadcast маленькой стороны. Идентично BHJ: Driver собирает меньшую таблицу через collect(), сериализует в byte[], транслирует через BlockManager. Если Spark не может определить, какая сторона меньше - он может broadcast обе или упасть с ошибкой.

Шаг 2 - Nested Loop. На каждом Executor'е для каждой строки из большой партиции выполняется полный проход по всем строкам broadcast-таблицы. Это не хеш-поиск - это последовательное сравнение. Для каждой пары строк вычисляется предикат (например, big.start_date <= small.end_date AND big.end_date >= small.start_date). При совпадении - строка эмитируется.

Сложность: O(N × M). Если большая таблица 10M строк, а маленькая 1000 строк - выполняется 10 миллиардов сравнений. Это катастрофа при любом реальном объёме.

Нет spill. Broadcast-таблица полностью в памяти, nested loop не может её спиллить. Если broadcast-таблица большая - OOM.

Когда BNLJ неизбежен¶

# Range join - нет equi-условия → BNLJ
df_events.join(
    df_time_ranges,
    (df_events.ts >= df_time_ranges.start_ts) &
    (df_events.ts <= df_time_ranges.end_ts)
)

# Условное соединение с !=
df_a.join(df_b, df_a.value != df_b.value)

# Cross join с фильтром (= BNLJ)
spark.sql("""
    SELECT /*+ SHUFFLE_REPLICATE_NL(r) */ *
    FROM events e JOIN ranges r
    ON e.ts BETWEEN r.start_ts AND r.end_ts
""")

Как митигировать BNLJ¶

При range join большой таблицы с маленькой таблицей интервалов - максимально фильтруйте ДО join:

# Плохо: BNLJ на 100M строк × 5000 интервалов
df_events.join(df_ranges, (df_events.ts >= df_ranges.s) & (df_events.ts <= df_ranges.e))

# Лучше: сначала отфильтровать события нужного периода
# и уменьшить df_ranges до активных интервалов
df_active_ranges = df_ranges.filter("end_ts >= current_timestamp()")
df_recent_events = df_events.filter("ts >= '2024-01-01'")
df_recent_events.join(broadcast(df_active_ranges),
    (df_recent_events.ts >= df_active_ranges.s) &
    (df_recent_events.ts <= df_active_ranges.e))

Для range join миллионов строк рассмотрите bucketing по времени (floor(ts / bucket_size)) как суррогатный equi-ключ.

5. Cartesian Product Join¶

Принцип: полное декартово произведение¶

Cartesian join создаёт все возможные пары строк из двух таблиц - без какого-либо условия соединения. Результат: N × M строк.

Физический алгоритм¶

Spark не может использовать хеш-партиционирование (нет ключа). Вместо этого:

1. Обе стороны читаются (с возможным broadcast меньшей стороны).
2. Каждый Executor, обрабатывающий партицию левой таблицы, получает полную копию правой таблицы (через broadcast или shuffle replication).
3. Для каждой строки левой партиции выполняется итерация по всей правой таблице - вложенный цикл без условия фильтрации.

# Требует явного разрешения (защита от случайного cross join)
spark.conf.set("spark.sql.crossJoin.enabled", "true")

# API crossJoin() - явный и не требует конфига
result = df_dates.crossJoin(df_products)

# Типичный use-case: grid для ML
dates = spark.range(30).selectExpr("date_add('2024-01-01', cast(id as int)) as date")
products = spark.createDataFrame([(1, "A"), (2, "B")], ["pid", "name"])
grid = dates.crossJoin(products)  # 30 × 2 = 60 строк

# SQL
spark.sql("""
    SELECT d.date, p.name
    FROM dates d CROSS JOIN products p
""")

Масштабирование: почему Cartesian опасен¶

Даже 100K × 1000 = 100M строк - это уже несколько ГБ данных. Используйте Cartesian только для контролируемых небольших датасетов (< 10K строк с каждой стороны).

Сравнительная таблица всех стратегий¶

AQE: адаптивное переключение стратегий в рантайме¶

Adaptive Query Execution (AQE) - ключевая оптимизация Spark 3.x, позволяющая менять стратегию JOIN во время выполнения, когда уже известна реальная статистика.

# Включить AQE (default: true в Spark 3.2+)
spark.conf.set("spark.sql.adaptive.enabled", "true")

Три оптимизации AQE¶

1. Динамический Broadcast Hash Join. Catalyst статически решил использовать SMJ (статистика показала большую таблицу). Во время Shuffle AQE видит реальный размер партиций. Если после Shuffle одна сторона оказалась < порога - AQE переключается на BHJ налету, без перезапуска задания.

spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "30m")

2. Coalesce shuffle partitions (слияние мелких партиций). После Shuffle 200 партиций → AQE смотрит реальные размеры. Если большинство партиций пустые или крошечные - объединяет их в меньшее число. Результат: меньше tasks, меньше overhead на Task scheduling.

spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

3. Skew Join Optimization. AQE обнаруживает «горячие» партиции (в X раз крупнее медианы) и автоматически дробит их на подпартиции, каждую обрабатывая отдельной задачей. Это устраняет skew без ручного salting.

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

Как AQE переключает стратегию: пример¶

Статическое планирование:         Рантайм (AQE):
orders JOIN customers              После Shuffle:
    ↓ (статистика неизвестна)          orders: 500 MB партиции
    SortMergeJoin ← план               customers: 8 MB итого
                                       ↓ AQE перепланирует
                                   BroadcastHashJoin ← реальное выполнение

Data Skew и Salting¶

Что такое skew и почему он ломает JOIN¶

Data skew - неравномерное распределение данных по партициям. При JOIN по полю user_id один «китовый» пользователь может иметь 10 миллионов транзакций, тогда как средний пользователь - 10. Результат: одна задача обрабатывает 10M строк, пока 199 задач уже завершили работу.

Ручной Salting (когда AQE skewJoin недостаточно)¶

Salting - добавление случайного суффикса к ключу для «размазывания» горячего ключа по нескольким партициям:

from pyspark.sql import functions as F

N_SALT = 10  # количество соляных значений

# ШАГ 1: Большая таблица - добавляем случайную соль
df_events_salted = df_events.withColumn(
    "salt_key",
    F.concat(F.col("user_id").cast("string"),
             F.lit("_"),
             (F.rand() * N_SALT).cast("int").cast("string"))
)

# ШАГ 2: Маленькая таблица - дублируем для каждого соляного значения
# explode(array(0,1,2,...,N-1)) создаёт N копий каждой строки
df_dim_salted = df_dim.withColumn(
    "salt_key",
    F.explode(F.array([F.lit(i) for i in range(N_SALT)]))
).withColumn(
    "salt_key",
    F.concat(F.col("user_id").cast("string"),
             F.lit("_"),
             F.col("salt_key").cast("string"))
)

# ШАГ 3: JOIN по salt_key
result = df_events_salted.join(df_dim_salted, "salt_key").drop("salt_key")

Что происходит: горячий ключ 99999 превращается в 10 ключей 99999_0, 99999_1, ..., 99999_9. Строки большой таблицы случайно распределяются по ним. Строки маленькой таблицы дублируются 10 раз. JOIN теперь распределён равномерно.

Цена: маленькая таблица увеличивается в N_SALT раз. Выбирайте N_SALT = 10–50 в зависимости от степени перекоса.

Bucketing: предустранение Shuffle¶

Bucketing - механизм предварительного распределения данных по файлам так, что при JOIN Shuffle не нужен вообще. Если обе таблицы bucketized по одному ключу с одинаковым числом бакетов - Spark читает совместимые бакеты напрямую.

# Создаём bucketized таблицы (один раз, при записи)
df_orders.write \
    .bucketBy(64, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("orders_bucketed")

df_customers.write \
    .bucketBy(64, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("customers_bucketed")

# JOIN без Shuffle: Spark читает бакет N из обеих таблиц → локальный SMJ
spark.table("orders_bucketed").join(
    spark.table("customers_bucketed"),
    "customer_id"
)

Проверяем в explain() - не должно быть Exchange:

== Physical Plan ==
Project [...]
+- SortMergeJoin [customer_id#0], [customer_id#1], Inner
   :- Sort [customer_id#0 ASC NULLS FIRST], false, 0   ← без Exchange
   +- Sort [customer_id#1 ASC NULLS FIRST], false, 0   ← без Exchange

Ограничения bucketing: число бакетов у обеих таблиц должно совпадать. Bucket key должен совпадать с join key. Таблицы должны быть Hive/managed (не внешние файлы). Изменить число бакетов после записи - только перезапись.

Join Hints: полная таблица¶

Hints переопределяют решение Catalyst. Использовать только когда у Catalyst плохая статистика или вы точно знаете размеры.

# Python API: hint применяется к одной стороне join
df_large.join(df_small.hint("broadcast"), "key")     # → BHJ
df_a.join(df_b.hint("merge"), "key")                  # → SMJ
df_big.join(df_med.hint("shuffle_hash"), "key")       # → SHJ
df_events.join(df_ranges.hint("shuffle_replicate_nl"),# → BNLJ
    condition)

-- SQL hints (сразу после SELECT):
SELECT /*+ BROADCAST(d)            */ * FROM facts f JOIN dims d ON f.id = d.id
SELECT /*+ MERGE(a, b)             */ * FROM a JOIN b ON a.k = b.k
SELECT /*+ SHUFFLE_HASH(s)         */ * FROM big b JOIN small s ON b.k = s.k
SELECT /*+ SHUFFLE_REPLICATE_NL(r) */ * FROM events e JOIN ranges r ON e.ts BETWEEN r.s AND r.e

Приоритет hints¶

Если hints конфликтуют (например, оба BROADCAST и MERGE) - Catalyst применяет приоритет:

1. BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL
2. Hint для меньшей стороны учитывается первым при BHJ.
3. Если hint физически невозможен (например, BROADCAST для 10 ГБ таблицы) - Catalyst игнорирует hint и выбирает следующую стратегию.

Чтение EXPLAIN: как понять, какая стратегия выбрана¶

explain() - главный инструмент диагностики JOIN-стратегии. Режим formatted показывает физический план с деталями.

df_result = df_orders.join(df_customers, "customer_id")
df_result.explain(mode="formatted")

Ключевые узлы в плане¶

== Physical Plan ==
BroadcastHashJoin [customer_id#0], [customer_id#1], Inner, BuildRight
:- Exchange hashpartitioning(customer_id#0, 200), ENSURE_REQUIREMENTS
:  +- Scan parquet orders [...]
+- BroadcastExchange HashedRelationBroadcastMode([customer_id#1], false)
   +- Scan parquet customers [...]

Разберём ключевые строки:

• BroadcastHashJoin - выбрана стратегия BHJ
• BuildRight - хеш-таблица строится из правой таблицы (customers)
• BroadcastExchange - правая таблица транслируется
• Exchange hashpartitioning у левой таблицы - левая таблица перемешивается (для совместимости партиций, иногда бывает)

# Пример SMJ
df_orders.join(df_customers.hint("merge"), "customer_id").explain("formatted")

# Ищем:
# SortMergeJoin
# Exchange hashpartitioning ← у обеих сторон
# Sort ← у обеих сторон

# Пример с AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
df_orders.join(df_customers, "customer_id").explain("formatted")
# Может показать AdaptiveSparkPlan с BroadcastHashJoin
# даже если статически планировался SMJ

Диагностика: как понять, что JOIN тормозит¶

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

Spark UI → вкладка SQL → нажать на запрос → открывается граф физического плана с метриками по каждому узлу. Это главный инструмент для диагностики проблем JOIN.

Что смотреть в графе плана:

• Duration у Exchange-узла - время, потраченное на Shuffle. Если это > 50% общего времени - Shuffle является узким местом.
• Shuffle Read/Write - объём данных, перемещённых по сети. Неожиданно большое значение = data explosion (дублирующиеся ключи или Cartesian).
• Rows Output у JOIN-узла - если намного больше, чем входные строки, значит ключи дублируются и JOIN производит Cartesian-like результат.
• Spill (Memory) / Spill (Disk) - ненулевые значения = памяти не хватает.

# Получить ссылку на Spark UI из кода
print(spark.sparkContext.uiWebUrl)
# → http://localhost:4040

# Включить детальные метрики SQL
spark.conf.set("spark.sql.ui.retainedExecutions", "100")

Метрики Spill: Disk vs RAM¶

Spill происходит, когда задача потребляет больше памяти, чем выделено в execution memory (часть от spark.executor.memory * spark.memory.fraction):

Метрика в Spark UI:
  Spill (Memory): 4.5 GB  ← сколько данных было in-memory перед сбросом
  Spill (Disk):   1.2 GB  ← сколько занял spill-файл (после сжатия)

Коэффициент Spill(Memory) / Spill(Disk) ≈ 3.75x - это коэффициент сжатия. Большой Spill (Memory) означает: задача получила слишком большую партицию и не смогла её обработать в RAM.

# Фиксы при Spill:

# 1. Увеличить число партиций (уменьшить размер каждой)
spark.conf.set("spark.sql.shuffle.partitions", "800")  # default 200

# 2. Увеличить execution memory fraction
spark.conf.set("spark.memory.fraction", "0.7")          # default 0.6
spark.conf.set("spark.memory.storageFraction", "0.4")  # default 0.5

# 3. Включить AQE coalesce (он не увеличит партиции, но уберёт пустые)
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64m")

Диагностика Data Skew в Spark UI¶

Skew-задачи видны на вкладке Stages в Spark UI. Нажмите на Stage с Shuffle Read - внизу страницы Timeline задач:

• Норма: все bars задач примерно одинаковой длины
• Skew: один bar в 10–100 раз длиннее остальных - это «straggler task»

В таблице задач столбец Duration: если медиана = 30 сек, а максимум = 45 мин - явный skew по одному ключу.

# Программная диагностика: найти горячие ключи
from pyspark.sql import functions as F

total_count = df.count()

df.groupBy("join_key") \
  .count() \
  .withColumn("pct", F.round(F.col("count") / total_count * 100, 2)) \
  .orderBy(F.col("count").desc()) \
  .show(10)

# Вывод:
# +--------+---------+-----+
# |join_key|    count|  pct|
# +--------+---------+-----+
# |   99999| 8000000 |80.00|   ← hot key: 80% всех строк
# |   12345|     150 | 0.00|
# |   67890|     200 | 0.00|
# +--------+---------+-----+

# Правило: ключ с > 1% всех строк при 1M+ записях - потенциальный hot key

Cost-Based Optimizer (CBO) и статистика¶

Зачем Catalyst нужна статистика¶

По умолчанию Catalyst строит план на основе приблизительного числа строк, часто вычисленного из размера файлов. Без точной статистики join-стратегии выбираются консервативно, порядок JOIN определяется по написанию запроса, а cardinality estimation бывает ошибочной в 10–100 раз.

CBO (Cost-Based Optimizer) включает гистограммы распределения данных по каждой колонке - это позволяет точнее оценить selectivity фильтров и размеры промежуточных результатов.

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

# Собрать базовую статистику (row count + total size)
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS")
spark.sql("ANALYZE TABLE customers COMPUTE STATISTICS")

# Собрать детальную статистику с гистограммами по колонкам
spark.sql("""
    ANALYZE TABLE orders
    COMPUTE STATISTICS FOR COLUMNS customer_id, amount, status, order_date
""")

# Проверить результат
spark.sql("DESCRIBE EXTENDED orders").show(50, truncate=False)
# Ищите строку: Statistics  X bytes, Y rows

Multi-way join reordering¶

При цепочке 4+ JOIN Catalyst с включённым CBO может переставить порядок JOIN так, чтобы промежуточные результаты были минимальны. Он оценивает cardinality каждого шага через гистограммы и выбирает порядок с наименьшей суммарной стоимостью:

# Собрать статистику по всем таблицам цепочки
tables = ["orders", "customers", "products", "regions", "categories"]
for t in tables:
    spark.sql(f"ANALYZE TABLE {t} COMPUTE STATISTICS FOR ALL COLUMNS")

# Запрос с 5 JOIN - Catalyst теперь переставит их оптимально
result = spark.sql("""
    SELECT
        o.order_id,
        c.name    AS customer_name,
        p.title   AS product_title,
        r.name    AS region_name,
        cat.name  AS category_name
    FROM orders o
    JOIN customers c   ON o.customer_id = c.id
    JOIN products p    ON o.product_id = p.id
    JOIN regions r     ON o.region_id = r.id
    JOIN categories cat ON p.category_id = cat.id
    WHERE c.country = 'RU'
      AND p.price > 1000
""")

# Без CBO: JOIN выполняется в порядке написания.
#   orders × customers → 100M строк → × products → 500M строк → ...
# С CBO: Catalyst сначала применяет фильтры (country='RU' → 10M строк,
#   price>1000 → 5M строк), затем строит дерево от малых к большим.
result.explain("formatted")

Dynamic Partition Pruning (DPP)¶

DPP - одна из самых мощных оптимизаций Spark 3.x для схемы «звезда». Она позволяет вообще не читать партиции fact-таблицы, которые заведомо не войдут в результат JOIN с dimension-таблицей.

Как работает DPP¶

Механика DPP - три шага:

1. Spark выполняет Scan dimension table с фильтром WHERE region_name = 'Russia' → получает набор {region_id: 1, 2, 3}.
2. Этот набор превращается в динамический фильтр-предикат для fact-таблицы: region_id IN (1, 2, 3).
3. При чтении fact-таблицы Spark применяет этот предикат как partition pruning - файлы из партиций region_id NOT IN (1, 2, 3) не открываются вообще. Экономия I/O: 97%.

Условия для срабатывания DPP:

• Fact-таблица должна быть партиционирована по join-ключу (partitionBy("region_id") при записи).
• JOIN должен использовать BroadcastHashJoin (DPP не работает с SMJ).
• Условие - equi-join по partition column.

# DPP включён по умолчанию в Spark 3.x
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")

# Fact-таблица партиционирована по join-ключу
df_events.write \
    .partitionBy("region_id") \
    .format("parquet") \
    .save("events_partitioned/")

# Dimension-таблица < порога broadcast → BHJ → DPP сработает
spark.sql("""
    SELECT e.event_id, e.user_id, r.region_name
    FROM events_partitioned e
    JOIN dim_regions r ON e.region_id = r.id
    WHERE r.region_name = 'Russia'
""")

# Проверяем в explain - должны увидеть DynamicPruning
df_result.explain(mode="formatted")
# PartitionFilters: [isnotnull(region_id#0),
#   dynamicpruningexpression(region_id#0 IN dynamicpruning#5)]
# ←  ^^^ это DPP! Spark передаёт динамический список partition values

Фильтрация и Column Pruning до JOIN¶

Predicate Pushdown: фильтруй до JOIN¶

Самый простой и мощный способ ускорить JOIN - уменьшить объём данных до него. Фильтрация сокращает Shuffle volume, размер промежуточных датасетов и нагрузку на сеть.

# ПЛОХО: фильтр ПОСЛЕ JOIN - сначала перемещаются все данные
result = df_orders \
    .join(df_customers, "customer_id") \
    .filter("country = 'RU' AND order_date >= '2024-01-01'")
# Catalyst часто делает pushdown автоматически, но не всегда - зависит от плана

# ХОРОШО: явный фильтр ДО JOIN - гарантированно меньше данных в Shuffle
df_orders_filtered = df_orders.filter("order_date >= '2024-01-01'")
df_customers_ru = df_customers.filter("country = 'RU'")

result = df_orders_filtered.join(df_customers_ru, "customer_id")

Особый случай: явная фильтрация NULL перед JOIN. Строки с NULL в join-ключе никогда не совпадут в INNER JOIN (NULL != NULL в SQL). Но без явного фильтра они участвуют в Shuffle и сортировке - тратят ресурсы впустую:

from pyspark.sql import functions as F

# NULL-строки попадают в Shuffle, но никогда не дают совпадений
df_orders.join(df_customers, "customer_id")  # NULL customer_id тратят Shuffle

# Правильно: явно отфильтровать NULL до JOIN
df_orders_clean = df_orders.filter(F.col("customer_id").isNotNull())
df_customers_clean = df_customers.filter(F.col("customer_id").isNotNull())
result = df_orders_clean.join(df_customers_clean, "customer_id")

Column Pruning: убирай ненужные колонки¶

Column Pruning уменьшает объём данных в Shuffle - по сети передаются только нужные колонки. Catalyst делает это автоматически при select() после join, но до Shuffle может не распространить pruning. Явный select() перед join гарантирует pruning на стадии Scan:

# ПЛОХО: 200-колоночная × 150-колоночная таблица = 350 колонок летят по сети
df_wide_orders.join(df_wide_customers, "customer_id")

# ХОРОШО: выбираем только нужное ДО JOIN
df_orders_slim = df_wide_orders.select(
    "order_id", "customer_id", "amount", "order_date", "status"
)
df_customers_slim = df_wide_customers.select(
    "customer_id", "name", "email", "country"
)

result = df_orders_slim.join(df_customers_slim, "customer_id")
# В Shuffle уходит 9 колонок вместо 350 → кратное сокращение сетевого трафика

Join order optimization¶

При цепочке JOIN важно расставлять их от самых селективных к наименее - сначала те, что отсекают больше строк:

# ПЛОХО: сначала JOIN с огромной таблицей без фильтра
df_events \
    .join(df_all_users, "user_id") \           # 100M × 50M → огромный промежуток
    .join(df_premium.filter("is_active"), "user_id")  # фильтр поздно

# ХОРОШО: сначала применяем наиболее селективный JOIN
df_premium_active = df_premium.filter("is_active")  # 5M строк вместо 50M

df_events \
    .join(broadcast(df_premium_active), "user_id") \  # broadcast 5M строк
    .join(df_all_users.select("user_id", "name", "email"), "user_id")
# Промежуточный датасет после 1-го JOIN: только события premium-пользователей

Repartition перед JOIN¶

Когда ручное repartition помогает¶

Если данные уже хранятся с партиционированием, несовместимым с join-ключом (например, по дате, а JOIN по user_id), явный repartition() по join-ключу может помочь Spark устранить один из двух Exchange:

# Данные записаны partitionBy("event_date"), а JOIN по user_id
# Без repartition: оба датасета пройдут Exchange → двойной Shuffle

# С repartition одной стороны по join-ключу:
df_events_repartitioned = df_events.repartition(200, "user_id")
result = df_events_repartitioned.join(df_users, "user_id")

# Проверяем: Exchange может исчезнуть у repartitioned стороны
result.explain("formatted")
# Если Spark видит совместимое партиционирование - Exchange не создаётся

Coalesce vs Repartition¶

# coalesce - без Shuffle, но партиции неравномерные (некоторые могут быть гигантскими)
df.coalesce(50)

# repartition - с Shuffle, равномерные партиции по hash
df.repartition(200)

# repartition по join-ключу - данные распределены как при Shuffle JOIN,
# если обе стороны repartitioned с одинаковым N → Exchange не нужен
df_orders_rep = df_orders.repartition(64, "customer_id")
df_customers_rep = df_customers.repartition(64, "customer_id")

# JOIN без Exchange (Spark видит совместимое партиционирование)
result = df_orders_rep.join(df_customers_rep, "customer_id")
result.explain("formatted")  # не должно быть Exchange ни у одной стороны

Narrow Transformation как альтернатива JOIN¶

Когда join можно заменить условной логикой¶

Для небольших статических справочников (10–100 значений), которые никогда не меняются, иногда выгоднее заменить JOIN на F.when() - условную логику внутри одной трансформации. Это narrow transformation: она выполняется параллельно внутри каждой партиции без какого-либо сетевого взаимодействия.

Диаграмма показывает ключевое различие: JOIN к справочнику требует хотя бы broadcast (collect → serialize → send), тогда как when() - это чистая вычислительная логика внутри каждого Executor'а без какого-либо движения данных по сети.

Практический пример: замена join к справочнику origin¶

from pyspark.sql import functions as F

# ПОДХОД 1: JOIN к справочнику (требует shuffle или broadcast)
df_origin_map = spark.createDataFrame([
    ("Ambulance", "Emergency"),
    ("Police",    "Emergency"),
    ("Person",    "Public"),
    ("Animal",    "Environmental"),
], ["origin_of_call", "category"])

result_join = df_rescues.join(broadcast(df_origin_map), "origin_of_call", "left")

# ПОДХОД 2: Narrow Transformation (нет shuffle, нет broadcast)
result_narrow = df_rescues.withColumn(
    "category",
    F.when(F.col("origin_of_call").isin("Ambulance", "Police"), "Emergency")
     .when(F.col("origin_of_call") == "Person", "Public")
     .when(F.col("origin_of_call") == "Animal", "Environmental")
     .otherwise("Other")
)

# Проверяем план - не должно быть ни Exchange, ни BroadcastExchange
result_narrow.explain("formatted")
# == Physical Plan ==
# *(1) Project [..., CASE WHEN ... END AS category]  ← только Project, нет Join!
# +- *(1) Scan parquet rescues [...]

Когда Narrow Transformation выгодна¶

Когда Narrow Transformation - анти-паттерн¶

# ПЛОХО: справочник с 1000+ значениями через when/otherwise
# - длинная цепочка условий трудна в обслуживании и тестировании
result = df.withColumn("region_name",
    F.when(F.col("region_code") == "MSK", "Москва")
     .when(F.col("region_code") == "SPB", "Санкт-Петербург")
     # ... ещё 85 регионов ...
     .otherwise("Неизвестно")
)
# ХОРОШО для 85 регионов: broadcast join к DataFrame - легче менять данные без изменения кода

# ХОРОШО: narrow transformation для <= ~20 статических значений
result = df.withColumn("priority",
    F.when(F.col("status") == "CRITICAL", 1)
     .when(F.col("status") == "HIGH",     2)
     .when(F.col("status") == "MEDIUM",   3)
     .otherwise(4)
)

Главное правило: Narrow Transformation оправдана только для маленьких, статических справочников. Если значения справочника меняются или их больше ~20–30 - используйте broadcast join: это надёжнее и проще в обслуживании.

Iterative Broadcast: горячие и холодные ключи¶

Проблема: один hot key ломает весь JOIN¶

Стандартный salting увеличивает маленькую таблицу в N_SALT раз. Если маленькая таблица уже 500 МБ, а нужен salt × 20 - это 10 ГБ, что не влезает в broadcast и перегружает SMJ. Решение - Iterative Broadcast: разделить данные на «горячие» (hot) и «холодные» (cold) ключи, обработать отдельными JOIN и объединить результаты.

from pyspark.sql import functions as F

# ШАГ 1: Найти горячие ключи (top N по частоте)
total = df_large.count()
key_counts = df_large.groupBy("join_key").count()

hot_keys = key_counts \
    .filter(F.col("count") > total * 0.001) \  # ключи с > 0.1% всех строк
    .select("join_key") \
    .rdd.flatMap(lambda x: x) \
    .collect()  # собрать на Driver - их обычно немного

# ШАГ 2: Разделить большую таблицу
df_hot = df_large.filter(F.col("join_key").isin(hot_keys))
df_cold = df_large.filter(~F.col("join_key").isin(hot_keys))

# ШАГ 3: Разделить маленькую таблицу
df_dim_hot = df_dim.filter(F.col("join_key").isin(hot_keys))
df_dim_cold = df_dim.filter(~F.col("join_key").isin(hot_keys))

# ШАГ 4: Горячие ключи - salting + broadcast
N_SALT = 20
df_hot_salted = df_hot.withColumn(
    "_sk",
    F.concat_ws("_",
        F.col("join_key").cast("string"),
        (F.rand() * N_SALT).cast("int").cast("string")
    )
)
df_dim_hot_salted = df_dim_hot \
    .withColumn("_arr", F.array([F.lit(i) for i in range(N_SALT)])) \
    .withColumn("_sv", F.explode("_arr")) \
    .withColumn("_sk", F.concat_ws("_", F.col("join_key").cast("string"), F.col("_sv").cast("string"))) \
    .drop("_arr", "_sv")

result_hot = df_hot_salted \
    .join(broadcast(df_dim_hot_salted), "_sk") \
    .drop("_sk")

# ШАГ 5: Холодные ключи - обычный broadcast (dim_cold мала, нет skew)
result_cold = df_cold.join(broadcast(df_dim_cold), "join_key")

# ШАГ 6: Объединить
result = result_hot.union(result_cold)

Преимущество: маленькая таблица дублируется в N_SALT раз только для горячих ключей (их обычно 1–5% от уникальных значений). Холодные ключи обрабатываются без дублирования.

Broadcast vs Side-loading (Python dict)¶

Когда broadcast DataFrame неэффективен¶

Стандартный BHJ через broadcast(df) работает на уровне Spark operators. Для Pandas UDF или обычных Python UDF broadcast DataFrame неэффективен - каждый вызов UDF делает дорогой row-by-row lookup через Spark API. Быстрее - broadcast Python dict и lookup напрямую в Python:

from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

# ШАГ 1: Собрать lookup-таблицу в Python dict на Driver
currency_rows = spark.table("exchange_rates") \
    .filter("rate_date = current_date()") \
    .select("currency_code", "rate_to_usd") \
    .collect()

rates_dict = {row.currency_code: float(row.rate_to_usd) for row in currency_rows}

# ШАГ 2: Broadcast Python dict через SparkContext (не через SQL!)
# Сериализуется через pickle, распределяется как BitTorrent
broadcast_rates = spark.sparkContext.broadcast(rates_dict)

# ШАГ 3: Pandas UDF читает broadcast dict локально на каждом Executor
@pandas_udf(DoubleType())
def convert_to_usd(currency: pd.Series, amount: pd.Series) -> pd.Series:
    rates = broadcast_rates.value  # O(1) - dict уже в памяти Executor
    return amount * currency.map(lambda c: rates.get(c, 1.0))

result = df_orders.withColumn(
    "amount_usd",
    convert_to_usd(F.col("currency"), F.col("amount"))
)
# Нет Shuffle, нет JOIN, нет Exchange - только векторизованный lookup в Pandas

Когда использовать broadcast DataFrame vs broadcast dict:

# Явно освободить broadcast переменную когда она больше не нужна
broadcast_rates.unpersist()

JOIN + Delta Lake оптимизация¶

Z-ORDER для join-heavy таблиц¶

Z-ORDER кластеризует файлы Delta Lake по значениям одной или нескольких колонок. При чтении с фильтром по этим колонкам срабатывает Data Skipping - файлы, чьи min/max статистики не пересекаются с фильтром, не открываются вообще. Для JOIN это значит: fact-таблица читается частично, только нужные файлы.

from delta.tables import DeltaTable

# Z-ORDER по join-ключу - выполняется один раз или по расписанию
DeltaTable.forPath(spark, "s3://datalake/events/") \
    .optimize() \
    .executeZOrderBy("customer_id")

# Теперь при JOIN по customer_id:
# Delta log хранит min/max customer_id для каждого файла
# При JOIN с фильтром на customers → DPP или явный фильтр
# → Spark читает только файлы с нужным диапазоном customer_id

spark.sql("""
    SELECT e.*, c.name, c.segment
    FROM delta.`s3://datalake/events/` e
    JOIN customers c ON e.customer_id = c.id
    WHERE c.segment = 'premium'
""")
# Без Z-ORDER: читаем 100% файлов events
# С Z-ORDER: Data Skipping пропускает ~70–90% файлов

Partition + Z-ORDER: комбинированная стратегия¶

Оптимальная стратегия для крупных fact-таблиц - двухуровневая фильтрация:

# Запись: partition по дате + Z-ORDER по join-ключу внутри каждой партиции
df_events.write \
    .format("delta") \
    .partitionBy("event_date") \
    .save("s3://datalake/events_optimized/")

# Оптимизация только нужных партиций (не перестраивать всё!)
DeltaTable.forPath(spark, "s3://datalake/events_optimized/") \
    .optimize() \
    .where("event_date >= '2024-01-01'") \    # только свежие партиции
    .executeZOrderBy("customer_id")

# При JOIN:
# Уровень 1: Partition Pruning по event_date → читаем только нужные даты
# Уровень 2: Data Skipping по Z-ORDER → внутри партиций читаем нужные файлы
# Уровень 3: DPP (если dim-таблица broadcast) → совместно с обоими уровнями

Liquid Clustering (Delta Lake 3.1+)¶

Liquid Clustering - альтернатива Z-ORDER + partitionBy без ручного управления партициями. Spark сам поддерживает оптимальный layout:

# Создание таблицы с Liquid Clustering
spark.sql("""
    CREATE TABLE events_lc
    USING delta
    CLUSTER BY (customer_id, event_date)
    AS SELECT * FROM events_raw
""")

# Incremental Clustering - запускается автоматически или вручную
spark.sql("OPTIMIZE events_lc")
# Spark перераспределяет только изменённые файлы, не всю таблицу
# Нет ограничений по числу partition columns (Z-ORDER деградирует при > 2 columns)

Anti-patterns JOIN pipeline¶

1. JOIN после explode без предварительной фильтрации¶

# ПЛОХО: explode ВЗРЫВАЕТ датасет в 50 раз, затем JOIN по огромному результату
df_orders \
    .withColumn("item", F.explode("items")) \   # 1M строк → 50M строк
    .join(df_products, "product_id")             # JOIN 50M строк - дорого

# ХОРОШО: фильтровать до explode
df_orders \
    .filter("status = 'completed'") \            # 1M → 200K строк
    .withColumn("item", F.explode("items")) \    # 200K → 10M строк
    .join(broadcast(df_products), "product_id")  # JOIN 10M строк - в 5 раз быстрее

2. SELECT * между широкими таблицами¶

# ПЛОХО: 200-колоночная × 150-колоночная = 350 колонок в Shuffle
df_wide_fact.join(df_wide_dim, "key")

# ХОРОШО: выбирать только нужное ДО JOIN
df_wide_fact.select("key", "metric_a", "metric_b") \
    .join(df_wide_dim.select("key", "dim_attr"), "key")
# Shuffle volume: 3 колонки вместо 350 → кратное ускорение

3. Множественные JOIN одной таблицы без кеширования¶

# ПЛОХО: df_dim читается и пересчитывается 3 раза
result = df_a.join(df_dim, "key") \
    .join(df_b, "other") \
    .join(df_dim, "key2")  # df_dim пересканируется!

# ХОРОШО: кешировать если используется несколько раз
df_dim.cache()
df_dim.count()  # материализовать кеш немедленно

result = df_a.join(df_dim, "key") \
    .join(df_b, "other") \
    .join(df_dim, "key2")

df_dim.unpersist()  # освободить кеш когда больше не нужен

4. Late filtering (фильтр после JOIN)¶

# ПЛОХО: весь JOIN выполняется, потом 95% строк отбрасывается
df_events \
    .join(df_users, "user_id") \
    .filter("created_at >= '2024-06-01'")  # фильтр ПОСЛЕ JOIN

# ХОРОШО: фильтр ДО JOIN - меньше данных в Shuffle
df_events.filter("created_at >= '2024-06-01'") \
    .join(df_users, "user_id")
# Примечание: Catalyst иногда делает pushdown автоматически,
# но явный код надёжнее и понятнее при code review

5. Случайный Cartesian Product¶

# КАТАСТРОФА: забыли условие JOIN → неявный Cartesian
df_a.join(df_b)  # N × M строк - часто зависает или OOM

# Защита: всегда указывайте условие явно
df_a.join(df_b, "key")          # equi-join
df_a.join(df_b, condition)      # non-equi join
df_a.crossJoin(df_b)            # явный cross join - намеренно

# Spark выдаёт AnalysisException при join() без условия если
# crossJoin.enabled = false (рекомендуется держать false в production)
spark.conf.set("spark.sql.crossJoin.enabled", "false")

Золотые правила production JOIN¶

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

Задание 1: Диагностика стратегии через EXPLAIN¶

Дан код:

df_sales = spark.read.parquet("sales/")          # 50 ГБ
df_products = spark.read.parquet("products/")    # 5 МБ
df_regions = spark.read.parquet("regions/")      # 200 МБ

result = df_sales \
    .join(df_products, "product_id") \
    .join(df_regions, "region_id")

Запустите explain(mode="formatted") и определите: какая стратегия выбрана для каждого JOIN? Почему? Что изменить, чтобы оба JOIN использовали BHJ?

from pyspark.sql.functions import broadcast

# Анализируем план
result.explain(mode="formatted")
# Ожидаемый результат:
# JOIN 1 (sales × products): BroadcastHashJoin → products (5 МБ < 10 МБ) транслируется
# JOIN 2 (× regions): SortMergeJoin → regions (200 МБ) превышает порог

# Для второго JOIN тоже BHJ - увеличиваем порог или явный broadcast:

# Вариант 1: увеличить порог (рискованно - потребляет память Executor'ов)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "250m")
result_v2 = df_sales.join(df_products, "product_id").join(df_regions, "region_id")
result_v2.explain(mode="formatted")
# Теперь оба JOIN → BroadcastHashJoin

# Вариант 2: явный hint (предпочтительнее - точный контроль)
result_v3 = df_sales \
    .join(broadcast(df_products), "product_id") \
    .join(broadcast(df_regions), "region_id")
result_v3.explain(mode="formatted")
# BroadcastHashJoin для обоих JOIN
# Суммарная память на Executor: 5 МБ + 200 МБ = 205 МБ × число Executor'ов
# При 50 Executor'ах: ~10 ГБ суммарно - убедитесь что хватает RAM

Задание 2: Устранение data skew через salting¶

Есть таблица events (1 миллиард строк) и user_profiles (5 млн строк). При JOIN по user_id одна задача обрабатывает 100x больше строк, чем остальные (пользователь-бот с миллионами событий). Напишите код с salting для равномерного распределения.

from pyspark.sql import functions as F

# Диагностика: смотрим распределение
df_events.groupBy("user_id").count().orderBy(F.col("count").desc()).show(5)
# user_id=99999: 80 000 000 строк  ← бот
# user_id=12345: 150 строк
# ...

N_SALT = 20  # 20 соляных значений для бота

# Большая таблица: случайная соль
df_events_salted = df_events.withColumn(
    "_salt_key",
    F.concat_ws("_",
        F.col("user_id").cast("string"),
        (F.rand() * N_SALT).cast("int").cast("string")
    )
)

# Маленькая таблица: дублируем N_SALT раз
df_profiles_salted = df_user_profiles.withColumn(
    "_salt_arr",
    F.array([F.lit(i) for i in range(N_SALT)])
).withColumn(
    "_salt_val", F.explode(F.col("_salt_arr"))
).withColumn(
    "_salt_key",
    F.concat_ws("_",
        F.col("user_id").cast("string"),
        F.col("_salt_val").cast("string")
    )
).drop("_salt_arr", "_salt_val")

# JOIN по составному ключу
result = df_events_salted.join(
    broadcast(df_profiles_salted),  # 5M × 20 = 100M строк → ~3 ГБ, можно broadcast
    "_salt_key"
).drop("_salt_key")

# Альтернатива: если profiles слишком большой для broadcast после дублирования - SMJ:
result_smj = df_events_salted.join(
    df_profiles_salted,  # SMJ без broadcast
    "_salt_key"
).drop("_salt_key")

# Проверяем план - не должно быть перекоса:
result.explain(mode="formatted")

Задание 3: Bucketing для ежедневных JOIN¶

Вы делаете ежедневный pipeline: orders JOIN customers. Обе таблицы большие (300 ГБ и 50 ГБ), join по customer_id. Сейчас 40% времени уходит на Shuffle. Настройте bucketing для устранения Shuffle.

# ШАГ 1: Записываем bucketized таблицы (один раз, или ежедневное обновление через append)

# Число бакетов: правило - sqrt(data_size_mb / target_file_size_mb)
# orders 300 ГБ: sqrt(300000 / 128) ≈ 48 → используем 64 (степень двойки)
# customers 50 ГБ: sqrt(50000 / 128) ≈ 20 → используем 64 (должно совпадать!)

N_BUCKETS = 64

df_orders.write \
    .mode("overwrite") \
    .bucketBy(N_BUCKETS, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("orders_bucketed")

df_customers.write \
    .mode("overwrite") \
    .bucketBy(N_BUCKETS, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("customers_bucketed")

# ШАГ 2: Читаем и выполняем JOIN
df_o = spark.table("orders_bucketed")
df_c = spark.table("customers_bucketed")

result = df_o.join(df_c, "customer_id")

# ШАГ 3: Проверяем план - Exchange не должен быть у обеих сторон
result.explain(mode="formatted")
# Ожидаем:
# SortMergeJoin [customer_id#0], [customer_id#1], Inner
# :- Sort [customer_id#0 ASC], false, 0       ← без Exchange! (уже отсортировано)
# +- Sort [customer_id#1 ASC], false, 0       ← без Exchange!

# Если видим Exchange - проверяем:
# 1. Число бакетов совпадает
# 2. Bucket key совпадает с join key
# 3. Таблицы managed (не external)
# 4. spark.sql.sources.bucketing.enabled = true (default)

Задание 4: AQE и диагностика переключения стратегий¶

Напишите код, который демонстрирует, как AQE переключает SMJ → BHJ в рантайме. Используйте explain(mode="extended") до и после выполнения.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "30m")
# Принудительно отключаем статический broadcast для демонстрации
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# Создаём тестовые данные: большая + относительно маленькая таблица
df_big = spark.range(10_000_000).selectExpr(
    "id as order_id",
    "cast(id % 100000 as int) as customer_id",
    "rand() * 1000 as amount"
)

df_small = spark.range(5_000).selectExpr(
    "id as customer_id",
    "concat('Customer_', id) as name"
)

df_result = df_big.join(df_small, "customer_id")

# Статический план (до выполнения): статистика неизвестна → SMJ
print("=== СТАТИЧЕСКИЙ ПЛАН ===")
df_result.explain(mode="formatted")
# Ожидаем: SortMergeJoin или AdaptiveSparkPlan (isFinalPlan=false)

# Выполняем - AQE анализирует реальные размеры после Shuffle
df_result.cache()
df_result.count()

# Финальный план (после выполнения): AQE переключился
print("=== ФИНАЛЬНЫЙ ПЛАН (после AQE) ===")
df_result.explain(mode="formatted")
# Ожидаем: BroadcastHashJoin (AQE увидел что df_small << 30 МБ)
# AdaptiveSparkPlan isFinalPlan=true
# BroadcastHashJoin [customer_id#0], [customer_id#1], Inner, BuildRight

# Ключевые индикаторы AQE в плане:
# AdaptiveSparkPlan isFinalPlan=true - план финализирован
# CustomShuffleReaderExec - AQE применил coalesce партиций
# BroadcastHashJoin вместо ожидаемого SortMergeJoin - AQE переключил стратегию

Задание 5: Комплексная оптимизация падающего pipeline¶

Кейс: ежедневный pipeline JOIN двух таблиц - transactions (500 ГБ, 2 млрд строк) и merchants (80 МБ, 400К строк). Pipeline либо падает с Executor Lost, либо работает 2 часа. Анализ Spark UI показывает:

• Stage 5 (Shuffle): одна задача обрабатывает 45 мин, остальные 199 - по 20 сек
• Spill (Memory): 12 GB на одной задаче
• Shuffle Read: 180 GB (ожидалось 15 GB - data explosion?)
• merchants таблица не проходит через BroadcastExchange

Пошагово: диагностируйте причины, примените исправления, проверьте через explain() и Spark UI метрики.

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

# ШАГ 1: ДИАГНОСТИКА - почему merchants не broadcast?
# Проверяем размер по статистике (Spark не знает реальный размер без ANALYZE)
df_merchants = spark.read.parquet("merchants/")
df_transactions = spark.read.parquet("transactions/")

# Spark UI показывает SortMergeJoin → статистика неизвестна или > 10 МБ порога
# Проверяем план:
df_transactions.join(df_merchants, "merchant_id").explain("formatted")
# Видим: SortMergeJoin + Exchange у обеих сторон
# Причина: autoBroadcastJoinThreshold = 10m, merchants = 80 MB > порога

# ШАГ 2: ДИАГНОСТИКА - data explosion (180 ГБ вместо 15 ГБ)
# Проверяем количество уникальных ключей vs дублирование
df_transactions.groupBy("merchant_id").count() \
    .orderBy(F.col("count").desc()).show(10)
# merchant_id=777: 1 800 000 000 строк (90% всех транзакций!) ← агрегатор/шлюз
# merchant_id=888: 150 000 000 строк
# Остальные: < 1000 строк каждый
# Это объясняет Spill и straggler: одна партиция = 90% всех данных

# ШАГ 3: ИСПРАВЛЕНИЕ ПРИЧИНЫ 1 - включить broadcast для merchants
# Вариант A: увеличить порог broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100m")  # 80 МБ < 100 МБ

# Вариант B: явный hint (надёжнее)
# Используем вариант B - более предсказуем

# ШАГ 4: ИСПРАВЛЕНИЕ ПРИЧИНЫ 2 - data skew через Iterative Broadcast
hot_merchants = ["777", "888"]  # найдены в ШАГ 2

# Разделяем транзакции
df_hot = df_transactions.filter(F.col("merchant_id").isin(hot_merchants))
df_cold = df_transactions.filter(~F.col("merchant_id").isin(hot_merchants))

df_merchants_hot = df_merchants.filter(F.col("merchant_id").isin(hot_merchants))
df_merchants_cold = df_merchants.filter(~F.col("merchant_id").isin(hot_merchants))

# Горячие ключи: salting N=10 + broadcast (merchants_hot × 10 = ~0.016 МБ)
N_SALT = 10
df_hot_salted = df_hot.withColumn(
    "_sk",
    F.concat_ws("_",
        F.col("merchant_id"),
        (F.rand() * N_SALT).cast("int").cast("string")
    )
)
df_merch_hot_salted = df_merchants_hot \
    .withColumn("_arr", F.array([F.lit(i) for i in range(N_SALT)])) \
    .withColumn("_sv", F.explode("_arr")) \
    .withColumn("_sk", F.concat_ws("_", F.col("merchant_id"), F.col("_sv").cast("string"))) \
    .drop("_arr", "_sv")

result_hot = df_hot_salted \
    .join(broadcast(df_merch_hot_salted), "_sk") \
    .drop("_sk")

# Холодные ключи: broadcast всей оставшейся merchants (80 МБ - крошки горячих)
result_cold = df_cold.join(broadcast(df_merchants_cold), "merchant_id")

# ШАГ 5: Финальный результат
result = result_hot.union(result_cold)

# ШАГ 6: Проверяем план - не должно быть Exchange ни в одной ветви
result_hot.explain("formatted")
# Ожидаем: BroadcastHashJoin (broadcast по _sk) - без Exchange
result_cold.explain("formatted")
# Ожидаем: BroadcastHashJoin (broadcast merchants_cold) - без Exchange

# ШАГ 7: Включить AQE и собрать статистику на будущее
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
# Для следующих запусков: AQE сам обнаружит skew и разобьёт партицию

spark.sql("ANALYZE TABLE transactions COMPUTE STATISTICS FOR COLUMNS merchant_id")
spark.sql("ANALYZE TABLE merchants COMPUTE STATISTICS")
# Теперь Catalyst будет знать реальный размер merchants и выберет BHJ автоматически

# ШАГ 8: Ожидаемый результат после оптимизации
# До:  2 часа, Executor Lost, Spill 12 GB, Shuffle 180 GB
# После: ~8 минут, без Spill, Shuffle ~5 GB (только маленькие таблицы в union)