Join стратегии: broadcast, sort-merge, shuffle hash и join hints

Физические алгоритмы join в Spark: Broadcast Hash Join, Sort-Merge Join, Shuffle Hash Join, join hints, AQE, data skew и bucketing для production pipeline.

core

Join стратегии: broadcast, sort-merge, shuffle hash и join hints¶

Join - самая дорогая операция в большинстве Spark pipeline. Именно join чаще всего становится узким местом: долгий shuffle, straggler-задачи из-за data skew, OOM на executor. При этом у Spark есть несколько физических алгоритмов join, каждый из которых оптимален при своих условиях.

В этом уроке мы разберём как Spark выбирает алгоритм, как прочитать explain plan чтобы понять что происходит, и как управлять выбором стратегии для максимальной производительности.

Анатомия join: что происходит в кластере¶

Join - это wide transformation: данные должны быть перегруппированы так, чтобы строки с одинаковым ключом оказались на одном executor. Это означает shuffle - самую дорогую операцию в распределённых системах.

Stage 2 (Exchange) - это сетевой трафик пропорциональный размеру обеих таблиц. Если таблицы весят сотни ГБ - это сотни ГБ по сети. Все последующие оптимизации направлены на уменьшение или полное исключение этого шага.

Типы join по семантике¶

Прежде чем говорить о стратегиях, напомним типы join по смыслу:

from pyspark.sql.functions import col, broadcast

# inner: только совпавшие строки (default)
orders.join(customers, "customer_id")
orders.join(customers, "customer_id", "inner")

# left: все строки orders, NULL для несовпавших customers
orders.join(customers, "customer_id", "left")
orders.join(customers, "customer_id", "left_outer")  # синоним

# right: все строки customers, NULL для несовпавших orders
orders.join(customers, "customer_id", "right")

# full: все строки обеих таблиц, NULL там где нет совпадения
orders.join(customers, "customer_id", "full")
orders.join(customers, "customer_id", "outer")  # синоним

# semi: строки orders где есть совпадение, без колонок customers
orders.join(customers, "customer_id", "left_semi")

# anti: строки orders где НЕТ совпадения (противоположность semi)
orders.join(customers, "customer_id", "left_anti")

# cross: декартово произведение (без условия)
small_a.join(small_b)  # опасно без фильтра - N*M строк

left_semi и left_anti - важные типы для data engineering: semi эквивалентен WHERE EXISTS (subquery), anti - WHERE NOT EXISTS. Они не добавляют колонки правой таблицы в результат, что снижает объём данных.

Broadcast Hash Join: самый быстрый путь¶

Broadcast Hash Join (BHJ) - стратегия когда маленькая таблица целиком копируется на каждый executor. Большая таблица остаётся на месте - никакого shuffle для неё:

Driver собирает маленькую таблицу в память, строит хэш-таблицу и рассылает её копию каждому executor. Затем каждый executor проверяет свои строки большой таблицы против локальной хэш-таблицы - без обмена данными между executor'ами.

Результат: Exchange (shuffle) для большой таблицы полностью исключён. Это в 3-10× быстрее Sort-Merge Join на практике.

Auto-broadcast threshold¶

Spark автоматически применяет BHJ если estimated size таблицы ≤ порога:

# Порог по умолчанию: 10 МБ (проверяется по статистике таблицы)
spark.conf.get("spark.sql.autoBroadcastJoinThreshold")  # "10485760" (10 MB)

# Увеличить для больших справочников
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(100 * 1024 * 1024))  # 100 MB

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

Важно: threshold проверяется по статистике каталога, а не по фактическому размеру. Если статистика устарела или отсутствует (нет ANALYZE TABLE), Spark может не применить BHJ даже для маленьких таблиц.

Риски и ограничения¶

Driver OOM: вся маленькая таблица собирается на Driver перед рассылкой. При 10 GB таблице и 4 GB Driver heap - крах.
Executor OOM: каждый executor хранит копию хэш-таблицы. При 100 executor'ах и 500 MB broadcast - 50 GB суммарного потребления памяти.
Не работает для cross join и non-equi join без дополнительных настроек.

# Проверить в explain - ищите BroadcastExchange и BroadcastHashJoin
orders.join(broadcast(products), "product_id").explain()
# == Physical Plan ==
# *(1) BroadcastHashJoin [product_id#0], [product_id#5], Inner, BuildRight
# :- *(1) Scan parquet [orders...]        ← без Exchange!
# +- BroadcastExchange HashedRelationBroadcastMode(...)
#    +- *(2) Scan parquet [products...]

Broadcast timeout: зависший Job¶

Особый класс проблем - когда broadcast join «зависает»: Job в состоянии Running, но в Spark UI нет ни одной активной Task.

Симптом:

Spark UI → Jobs: Job 5 Running... (уже 20 минут)
Spark UI → Stages → Stage 3: 0/200 tasks running, 0 completed
Driver logs:
  WARN TaskSetManager: No active tasks running...
  ERROR SparkContext: Exception in SparkContext.broadcast
  java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]

Причина: Driver собрал таблицу в память и пытается разослать её на все executor'ы через BroadcastExchange. Если таблица оказалась значительно больше ожидаемой (например, статистика устарела и таблица не 10 МБ, а 5 ГБ), рассылка занимает десятки минут.

Диагностика:

# Проверить реальный размер таблицы до broadcast
from pyspark.sql.functions import col

# Посмотреть оценку vs реальность
df.explain("cost")
# Statistics(sizeInBytes=10.0 MiB)  ← оценка Spark
# Если реальный размер другой - нужен ANALYZE TABLE

# Проверить количество строк и размер явно
print(f"Count: {df.count()}")
df.write.mode("overwrite").parquet("/tmp/size_check")
# → du -sh /tmp/size_check

Решения:

# 1. Убрать broadcast() hint если таблица большая
#    Вместо:
result = big_df.join(broadcast(also_big_df), "key")
#    Использовать:
result = big_df.join(also_big_df, "key")  # Spark сам выберет SortMergeJoin

# 2. Увеличить timeout если broadcast обоснован (большой кластер, медленная сеть)
spark.conf.set("spark.sql.broadcastTimeout", "600")  # 600 сек (по умолч. 300)

# 3. Отключить auto-broadcast для конкретного запроса
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# 4. Обновить статистику чтобы Spark не делал broadcast по ошибке
spark.sql("ANALYZE TABLE my_table COMPUTE STATISTICS")

Sort-Merge Join: стандарт для больших таблиц¶

Sort-Merge Join (SMJ) - алгоритм по умолчанию для соединения двух больших датасетов. Он надёжен, предсказуем и не требует помещать данные целиком в память executor:

Алгоритм в трёх шагах:

Exchange: обе таблицы перемешиваются по ключу - hash(join_key) % num_partitions. Строки с одинаковым ключом попадают в одну партицию одного executor.
Sort: каждый executor сортирует свою часть данных обеих таблиц по join-ключу. Это O(N log N) и может вызвать spill на диск.
Merge: линейный проход O(N+M) по двум отсортированным потокам - как merge в merge sort.

Стоимость и настройка¶

Узкое место SMJ - количество shuffle-партиций. По умолчанию spark.sql.shuffle.partitions = 200. Если данных 100 ГБ - 200 партиций по 500 МБ это нормально. Если 1 ТБ - нужно больше партиций:

# Рассчитайте: target partition size ~ 128-256 MB
# 1 TB / 200 MB = 5000 партиций
spark.conf.set("spark.sql.shuffle.partitions", "5000")

# Spark 3.x с AQE может делать это автоматически (coalescing)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.targetPostShuffleInputSize", "256mb")

Spill: сброс на диск¶

Если отсортированные данные не помещаются в executor memory - Spark сбрасывает часть на диск (spill). Это в 10-100× медленнее чем работа с RAM. Признаки spill в Spark UI: высокий Spill (disk) в Stage metrics.

# Увеличить executor memory fraction для sort buffer
spark.conf.set("spark.memory.fraction", "0.8")          # доля heap под Spark
spark.conf.set("spark.memory.storageFraction", "0.3")   # доля под cache vs execution

Shuffle Hash Join: хэш в памяти без сортировки¶

Shuffle Hash Join (SHJ) - средний вариант между BHJ и SMJ. Данные shuffle-ятся как в SMJ, но вместо сортировки + merge на каждом executor строится хэш-таблица из меньшей таблицы:

# SHJ включается когда:
# 1. Одна сторона намного меньше другой, но слишком велика для broadcast
# 2. spark.sql.join.preferSortMergeJoin = false
# 3. Данные не имеют смысла сортировать (нет orderBy downstream)

spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")

Преимущество перед SMJ: нет стоимости сортировки (O(N) vs O(N log N)). Недостаток: вся меньшая партиция должна помещаться в память executor. При skew одного ключа - OOM.

Spark 3.x с AQE предпочитает SMJ по умолчанию из-за его предсказуемости при spill.

Broadcast Nested Loop Join: опасный тип¶

Если условие join не является equi-join (a.col == b.col), Spark не может использовать хэш или sort-merge. Применяется Broadcast Nested Loop Join (BNLJ) - O(N×M) по сложности:

# Non-equi join: range condition
orders.join(
    promotions,
    (col("order_date") >= col("promo_start")) &
    (col("order_date") <= col("promo_end"))
)
# → BroadcastNestedLoopJoin - если promotions маленькая
# → CartesianProduct если обе большие → КАТАСТРОФА

# ANTI-PATTERN: cross join без filter
df_a.join(df_b)  # 1M × 1M = 1 трлн строк!

Для range join на больших таблицах рассмотрите: bucketing по времени, IntervalTree join (spark-rangejoin библиотеки), или переформулирование в equi-join через date_trunc.

Join Hints: берём управление на себя¶

Join hints позволяют переопределить выбор Catalyst. Используйте когда знаете о данных больше оптимизатора:

from pyspark.sql.functions import broadcast

# Способ 1: функция broadcast() - самый идиоматичный
result = orders.join(broadcast(stores), "store_id")

# Способ 2: hint через метод .hint()
result = orders.join(stores.hint("broadcast"), "store_id")

# Способ 3: SQL hint в selectExpr / spark.sql
result = spark.sql("""
    SELECT /*+ BROADCAST(s) */ o.*, s.store_name
    FROM orders o
    JOIN stores s ON o.store_id = s.store_id
""")

# Другие hints
orders.join(large_dim.hint("merge"),          "id")  # принудительный SMJ
orders.join(medium_table.hint("shuffle_hash"), "id")  # принудительный SHJ

Когда hints - это зло¶

Hints переопределяют статистику и могут дать плохой план при изменении данных:

# Опасно: broadcast таблицы которая выросла с 5 MB до 5 GB
# Spark попытается broadcast 5 GB → Driver OOM
result = orders.join(broadcast(fast_growing_table), "id")

# Лучше: доверьте порог AQE и укажите максимальный размер
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(200 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", str(200 * 1024 * 1024))

Правило: используйте broadcast() для стабильно маленьких справочников (currencies, regions, status codes). Для таблиц с непредсказуемым ростом - настройте threshold и доверьте AQE.

Adaptive Query Execution (AQE)¶

AQE - механизм Spark 3.x, который меняет план выполнения во время runtime на основе реальной статистики, а не оценок:

AQE включает три ключевые оптимизации:

1. Dynamic join strategy switching: если после первого shuffle правая сторона оказалась меньше порога - AQE переключает SMJ на BHJ без перезапуска запроса.

2. Coalescing shuffle partitions: если в 200 партициях данные неравномерны (100 маленьких + 5 больших) - AQE объединяет маленькие партиции.

3. Skew join optimization: AQE обнаруживает горячие партиции и разбивает их на подпартиции.

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

# AQE-специфичный порог для dynamic broadcast (может быть > autoBroadcastJoinThreshold)
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "64mb")

# Порог для определения skewed партиции
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256mb")

Data Skew: главная угроза производительности¶

Data skew в join - когда большинство строк имеют одинаковый join-ключ. Один executor получает 90% данных, остальные простаивают. Это straggler-задача - весь job ждёт одну медленную задачу:

# Диагностика skew: посмотреть распределение ключей
orders.groupBy("customer_id") \
    .count() \
    .orderBy(col("count").desc()) \
    .show(10)
# customer_id=NULL       → 5_000_000 строк  ← NULL-skew
# customer_id=999        → 2_000_000 строк  ← hot key
# customer_id=1          → 150 строк

Salting: ручная защита от skew¶

Если AQE skew join не справляется (или используется Spark < 3.0), применяйте ручное salting:

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

N_SALT = 10  # количество salt-партиций

# Левая таблица: добавляем случайный суффикс к горячему ключу
orders_salted = orders.withColumn(
    "salted_key",
    concat(
        col("customer_id").cast("string"),
        lit("_"),
        (rand() * N_SALT).cast("int").cast("string")
    )
)

# Правая таблица: дублируем строки с горячим ключом для всех salt значений
from pyspark.sql.functions import explode, array

customers_salted = customers.withColumn(
    "salt_range",
    array([lit(i) for i in range(N_SALT)])
).withColumn("salt", explode(col("salt_range"))) \
 .withColumn(
    "salted_key",
    concat(col("customer_id").cast("string"), lit("_"), col("salt").cast("string"))
 ).drop("salt_range", "salt")

# Join по salted_key - горячий ключ теперь разбит на N_SALT партиций
result = orders_salted.join(customers_salted, "salted_key") \
    .drop("salted_key")

Важно: salting применяется только к горячим ключам. Для всех остальных - overhead без пользы. В реальных системах: определяйте топ-N горячих ключей заранее и применяйте salting только к ним.

NULL как источник skew¶

NULL в join-ключе - частая причина skew. Все NULL собираются в одну партицию (hash(NULL) - фиксированное значение):

# Исключить NULL из join (они всё равно не совпадают в inner join)
orders_filtered = orders.filter(col("customer_id").isNotNull())

# Или: заменить NULL на уникальный sentinel перед join
# чтобы NULL не собирались в одну партицию
from pyspark.sql.functions import monotonically_increasing_id, when

orders_safe = orders.withColumn(
    "customer_id_safe",
    when(col("customer_id").isNull(),
         concat(lit("__null_"), monotonically_increasing_id().cast("string"))
    ).otherwise(col("customer_id").cast("string"))
)

Bucketing: join без shuffle¶

Если таблицы заранее сохранены с одинаковым bucketing (колонка, количество buckets) - join между ними не требует Exchange. Данные уже лежат в нужном разбиении:

# Запись с bucketing
orders.write \
    .bucketBy(200, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("orders_bucketed")

customers.write \
    .bucketBy(200, "customer_id") \
    .sortBy("customer_id") \
    .saveAsTable("customers_bucketed")

# Join без shuffle!
orders_b = spark.table("orders_bucketed")
customers_b = spark.table("customers_bucketed")

result = orders_b.join(customers_b, "customer_id")
result.explain()
# == Physical Plan ==
# *(3) SortMergeJoin [customer_id#0L], [customer_id#5L], Inner
# :- *(1) Sort ...                  ← нет Exchange перед Sort!
# +- *(2) Sort ...                  ← нет Exchange перед Sort!

Bucketing эффективен для регулярно выполняемых join одних и тех же таблиц. Первоначальная запись с bucketing дороже, но каждый последующий join экономит shuffle. Обязательное условие: обе стороны должны иметь одинаковое количество buckets и тот же join-ключ.

Читаем explain plan для join¶

Умение читать explain план - ключевой навык для оптимизации join:

result = (
    orders
    .filter(col("status") == "COMPLETED")
    .join(broadcast(stores), "store_id")
    .join(customers, "customer_id")
)
result.explain("formatted")

Что искать в плане:

Узел в плане	Что означает
`BroadcastHashJoin`	один из join без shuffle для большой таблицы
`SortMergeJoin`	стандартный join с двумя Exchange
`ShuffleHashJoin`	хэш-join после shuffle
`BroadcastNestedLoopJoin`	non-equi join, O(N×M) - опасно
`Exchange hashpartitioning(key, N)`	shuffle - дорогая сетевая операция
`BroadcastExchange`	рассылка маленькой таблицы всем executor
`*(N)` перед узлом	WholeStageCodeGen включён - эффективно

== Physical Plan ==
*(5) Project [order_id#0, amount#1, store_name#10, customer_name#20]
+- *(5) BroadcastHashJoin [customer_id#2], [customer_id#15], Inner, BuildRight
   :- *(5) Project [order_id#0, amount#1, customer_id#2, store_name#10]
   :  +- *(5) BroadcastHashJoin [store_id#3], [store_id#8], Inner, BuildRight
   :     :- *(5) Filter (status#4 = COMPLETED)
   :     :  +- *(5) Scan parquet orders[...]  ← только один Scan, нет Exchange
   :     +- BroadcastExchange HashedRelation...
   :        +- *(4) Scan parquet stores[...]   ← broadcast, нет Exchange
   +- BroadcastExchange HashedRelation...
      +- *(3) Scan parquet customers[...]       ← broadcast, нет Exchange

В этом плане три таблицы, ни одного Exchange - идеальный случай когда обе маленькие таблицы broadcast-ятся.

Практика¶

Задание 1: Анализ стратегий через explain¶

Есть три таблицы: transactions (1 ТБ), stores (50 МБ), products (500 МБ). Создайте план и найдите какой join strategy выбрал Spark:

transactions = spark.read.parquet("/data/silver/transactions/")
stores       = spark.read.parquet("/data/silver/stores/")
products     = spark.read.parquet("/data/silver/products/")

result = (
    transactions
    .filter(col("status") == "COMPLETED")
    .join(stores,   "store_id")
    .join(products, "product_id")
)
result.explain("formatted")

Задачи:

Найти в плане Exchange-узлы - сколько их и для каких join
Принудительно broadcast stores и проверить что Exchange исчез
Установить порог 600 МБ и проверить что products тоже стал broadcast

Решение

from pyspark.sql.functions import broadcast, col

# 1. Без hint: stores (50 MB) может автоматически broadcast, products (500 MB) - нет
# В плане: один BroadcastHashJoin для stores, SortMergeJoin для products
# Видим: Exchange hashpartitioning для transactions + products join

# 2. Принудительный broadcast stores
result_v2 = (
    transactions
    .filter(col("status") == "COMPLETED")
    .join(broadcast(stores), "store_id")   # ← явный broadcast
    .join(products, "product_id")
)
result_v2.explain()
# Теперь: BroadcastHashJoin для stores (BroadcastExchange без Exchange для transactions)
# Но products join всё равно SortMergeJoin с Exchange

# 3. Увеличить порог до 600 MB - products тоже попадает под auto-broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(600 * 1024 * 1024))
result_v3 = (
    transactions
    .filter(col("status") == "COMPLETED")
    .join(stores, "store_id")
    .join(products, "product_id")
)
result_v3.explain()
# Теперь: два BroadcastHashJoin, ни одного Exchange для transactions
# Transactions читается только один раз без перемещения данных!

# ВАЖНО: вернуть порог к разумному значению после эксперимента
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(100 * 1024 * 1024))

В explain("formatted") узлы без Exchange перед ними - это признак broadcast join. Если видите Exchange hashpartitioning(store_id#5, 200) - эта таблица участвует в shuffle.

Задание 2: Диагностика и лечение data skew¶

Создайте синтетический датасет с сильным перекосом ключа и примените salting:

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

# Транзакции: 90% с customer_id=1 (горячий ключ)
transactions = spark.range(100_000).select(
    col("id").alias("txn_id"),
    # 90% строк идут к customer_id=1
    (col("id") % 10 != 0).cast("int").alias("_is_hot"),
).withColumn(
    "customer_id",
    when(col("_is_hot") == 1, lit(1))
    .otherwise((col("txn_id") % 1000 + 2).cast("long"))
).drop("_is_hot")

customers = spark.range(1, 1002).select(
    col("id").alias("customer_id"),
    concat(lit("Customer_"), col("id")).alias("name"),
)

Нужно: выполнить join без salting, посмотреть на распределение партиций (через explain или подсчёт), затем применить salting и сравнить.

Решение

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

# Сначала: наивный join - customer_id=1 займёт одну огромную партицию
result_naive = transactions.join(customers, "customer_id")
# В Spark UI: одна задача Stage обрабатывает 90k строк, остальные по ~100

# Проверка распределения по партициям
transactions.groupBy("customer_id") \
    .count() \
    .orderBy(col("count").desc()) \
    .show(5)
# +----------+------+
# |customer_id| count|
# +----------+------+
# |         1 | 90000|  ← горячий ключ
# |       500 |   100|
# ...

# Salting: разбиваем горячий ключ на N партиций
N_SALT = 10

# Левая сторона: случайный salt
txn_salted = transactions.withColumn(
    "salt", floor(rand() * N_SALT).cast("int")
).withColumn(
    "salted_key",
    concat(col("customer_id").cast("string"), lit("_"), col("salt").cast("string"))
)

# Правая сторона: дублировать каждую строку N_SALT раз
cust_salted = customers.withColumn(
    "salts", array([lit(i) for i in range(N_SALT)])
).withColumn(
    "salt", explode(col("salts"))
).withColumn(
    "salted_key",
    concat(col("customer_id").cast("string"), lit("_"), col("salt").cast("string"))
).drop("salts", "salt")

result_salted = txn_salted.join(cust_salted, "salted_key") \
    .drop("salted_key", "salt", "customer_id_salted")

# Распределение теперь равномерное: customer_id=1 разбит на 10 партиций
# каждая содержит ~9000 строк вместо одной с 90000

После salting горячий ключ customer_id=1 создаёт 10 join-партиций вместо одной. Цена: правая таблица customers увеличивается в 10× по объёму (каждая строка дублируется). Salting выгоден когда горячих ключей мало, а правая таблица небольшая.

Задание 3: Влияние порядка join и ранней фильтрации¶

Покажите разницу в планах: join до фильтрации vs фильтрация до join:

orders    = spark.read.parquet("/data/silver/orders/")
customers = spark.read.parquet("/data/silver/customers/")
products  = spark.read.parquet("/data/silver/products/")

Решение

from pyspark.sql.functions import col, broadcast

# АНТИПАТТЕРН: join сначала, фильтр потом - shuffle всех данных
result_bad = (
    orders
    .join(customers, "customer_id")  # shuffle 1 TB orders + 50 GB customers
    .join(products,  "product_id")   # shuffle результата + products
    .filter(col("status") == "COMPLETED")  # фильтр после join!
    .filter(col("region") == "RU")
)
result_bad.explain()
# Exchange появляется перед фильтрацией: shuffle всего датасета

# ПРАВИЛЬНО: фильтр как можно раньше - меньше данных в shuffle
result_good = (
    orders
    .filter(col("status") == "COMPLETED")  # убрать 70% строк ДО join
    .join(broadcast(customers.filter(col("region") == "RU")), "customer_id")
    .join(products, "product_id")
)
result_good.explain()
# После первого фильтра: orders уменьшился с 1TB до ~300GB
# customers broadcast (50 GB * 0.3 region = ~15 GB - всё равно нужен порог)

# Подсказка: проверить что Catalyst сам делает filter pushdown
# В explain "extended" → Optimized Logical Plan
# Catalyst обычно поднимает фильтры вверх по плану автоматически,
# но явная ранняя фильтрация в коде помогает читаемости и
# гарантирует уменьшение данных при любой версии оптимизатора

Catalyst обычно поднимает фильтры автоматически (predicate pushdown), но явная ранняя фильтрация в коде гарантирует меньший объём данных для shuffle и делает намерения понятными для следующего разработчика.

Задание 4: AQE в действии¶

Сравните план с включённым и отключённым AQE для join с неравномерным размером после фильтрации:

transactions = spark.read.parquet("/data/silver/transactions/")

# Таблица references_large кажется Catalyst большой (500 МБ по статистике),
# но после фильтрации по region = 'RU' остаётся 8 МБ
references_large = spark.read.parquet("/data/silver/references/")

Решение

# Без AQE: Catalyst видит 500 МБ → планирует SortMergeJoin
spark.conf.set("spark.sql.adaptive.enabled", "false")
result_no_aqe = transactions.join(
    references_large.filter(col("region") == "RU"),
    "ref_id"
)
result_no_aqe.explain()
# == Physical Plan ==
# SortMergeJoin ← Catalyst не знает о реальном размере после фильтра
# Exchange hashpartitioning(ref_id, 200) ← shuffle обеих таблиц

# С AQE: после выполнения фильтра Spark видит реальный размер 8 МБ
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "64mb")

result_aqe = transactions.join(
    references_large.filter(col("region") == "RU"),
    "ref_id"
)
result_aqe.explain()
# == Physical Plan (после AQE перепланирования) ==
# BroadcastHashJoin ← AQE увидел 8 МБ → переключил на broadcast!
# НЕТ Exchange для transactions ← огромная экономия

# Разница в Spark UI:
# Без AQE: 2 Stage с Exchange, время ~10 мин
# С AQE:   1 Stage с BroadcastExchange, время ~2 мин

AQE перепланирует join после завершения первого Stage. Это особенно ценно когда статистика каталога устарела или таблица сильно фильтруется перед join. Включите AQE в production и установите autoBroadcastJoinThreshold для AQE выше чем базовый порог - AQE имеет реальную статистику и ошибётся реже.

Сравнение стратегий join¶

	Broadcast Hash Join	Sort-Merge Join	Shuffle Hash Join	Broadcast Nested Loop
Когда Spark выбирает	Одна сторона ≤ `autoBroadcastJoinThreshold` (10 МБ) или есть `broadcast()` hint	По умолчанию для двух больших таблиц	`preferSortMergeJoin=false` + одна сторона значительно меньше	non-equi join или cross join
Shuffle	Нет (Exchange исключён)	Обе таблицы (2 Exchange)	Обе таблицы (2 Exchange)	Нет (broadcast меньшей) или full cartesian
Сортировка	Нет	Да (TimSort внутри каждой партиции)	Нет	Нет
Требование к памяти	Вся маленькая таблица на каждом executor	Буфер сортировки (spill-safe)	Меньшая партиция в памяти executor	Вся меньшая таблица на каждом executor
Алгоритм сложность	O(N) probe по хэш-таблице	O(N log N) sort + O(N) merge	O(N) probe по хэш-таблице	O(N × M) вложенные циклы
Spill-safe	- (нет shuffle)	Да (ExternalSorter)	Нет (OOM при skew)	-
Equi-join	Обязателен	Обязателен	Обязателен	Не нужен (range, cross)
Физический план	`BroadcastHashJoin` + `BroadcastExchange`	`SortMergeJoin` + `Exchange hashpartitioning`	`ShuffledHashJoin` + `Exchange hashpartitioning`	`BroadcastNestedLoopJoin`
Типичное ускорение vs SMJ	3–10×	baseline	1.2–2× (нет sort)	0.001–0.1× (катастрофа)
Когда использовать	Справочники: города, валюты, статусы	Два больших датасета, skew-safe	Средние таблицы без sort downstream	Только если нет альтернативы
Hint	`broadcast(df)` или `.hint("broadcast")`	`.hint("merge")`	`.hint("shuffle_hash")`	-

Как Spark выбирает стратегию¶

Чек-лист для production join¶

Проверка	Инструмент
Размер правой таблицы < 200 МБ	`spark.sql.autoBroadcastJoinThreshold` + `broadcast()`
Есть фильтрация до join	`explain("extended")` → Optimized Logical Plan
NULL в join-ключе	`.filter(col("key").isNotNull())` перед join
Data skew (топ ключ > 1 ГБ)	AQE skew join или salting
Повторные join одних таблиц	Bucketing при записи
Много маленьких shuffle-партиций	AQE coalescing или уменьшить shuffle.partitions
Нет BroadcastExchange в плане	Проверить statistic (`ANALYZE TABLE`), увеличить threshold
non-equi join	Переформулировать как equi-join или принять cost