Что такое broadcast join в Spark и когда его использовать?

Broadcast Join в Apache Spark

Broadcast join — это одна из самых мощных оптимизаций в Spark для повышения производительности объединения таблиц. Это позволяет избежать дорогостоящего shuffle и значительно ускорить обработку данных. Разберу детально, как это работает и когда его применять.

Что такое broadcast join?

Определение: Broadcast join — это стратегия объединения, при которой меньшая таблица полностью копируется на каждый worker-узел, а затем объединяется с большей таблицей без shuffle операций.

Как это работает

Без broadcast join (обычный shuffle join):

Таблица A (1 TB):    Таблица B (100 GB):
Partition 0          Partition 0
Partition 1          Partition 1
Partition 2          Partition 2
...

Shuffle phase: Данные А и В перетасовываются по сети
(ОЧЕНЬ МЕДЛЕННО!)

Результат: объединённые партиции

С broadcast join:

Таблица A (1 TB):         Таблица B (100 GB):
Partition 0               Partition 0
Partition 1      ------->Partition 1  <- загружается полностью
Partition 2               Partition 2
...

Broadcast phase: Таблица B полностью отправляется на все ноды
(БЫСТРО! Потому что B маленькая)

Local join: На каждой ноде локально объединяем
A partition с полной таблицей B (в памяти)

Результат: объединённые партиции (БЕЗ SHUFFLE!)

Пример кода

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("broadcast_join").getOrCreate()

# Большая таблица (факты)
fact_sales = spark.read.parquet("s3://data/sales")  # 500 GB
fact_sales.count()  # Миллиарды строк

# Маленькая таблица (справочник)
dim_products = spark.read.parquet("s3://data/products")  # 50 MB
dim_products.count()  # Тысячи строк

# Способ 1: Явный broadcast
result = fact_sales.join(
    broadcast(dim_products),  # <- Явно отправляем на все ноды
    on="product_id",
    how="inner"
)

# Способ 2: Автоматический broadcast (Spark сам решает)
result = fact_sales.join(
    dim_products,
    on="product_id",
    how="inner"
)
# Spark автоматически broadcast-ит если dim_products < 100 MB

Архитектура broadcast join

Этап 1: Сбор маленькой таблицы на driver ноде

# На driver ноде маленькая таблица собирается полностью в памяти
# Это безопасно, потому что размер маленький

dim_products_data = dim_products.collect()  # собрали в памяти
# [(product_id=1, name="Phone", price=1000),
#  (product_id=2, name="Laptop", price=2000),
#  ...]

Этап 2: Отправка на все executor ноды

Driver node:
  {
    product_id=1 -> {name: "Phone", price: 1000},
    product_id=2 -> {name: "Laptop", price: 2000},
    ...
  }
  |
  ├──> Executor 0
  ├──> Executor 1
  ├──> Executor 2
  └──> Executor N

Каждый executor получает полную копию таблицы в памяти

Этап 3: Локальное объединение на каждом executor

# На каждом executor ноде:
# Для каждой строки fact_sales:
# - Ищем соответствующую строку в памяти (в hash-table)
# - Объединяем локально (БЕЗ SHUFFLE!)

# Очень быстро, потому что поиск в памяти O(1)

Когда использовать broadcast join

✅ Используй broadcast join если:

1. Одна из таблиц маленькая (< 100 MB, обычно < 1 GB)

   # Справочник категорий
   dim_categories = spark.read.parquet("s3://dim/categories")  # 5 MB
   
   fact_sales = spark.read.parquet("s3://fact/sales")  # 500 GB
   
   result = fact_sales.join(
       broadcast(dim_categories),
       on="category_id"
   )
   

2. Часто объединяешь одну большую таблицу с несколькими маленькими

   # Несколько broadcast join-ов подряд
   result = (
       fact_sales
       .join(broadcast(dim_products), on="product_id")
       .join(broadcast(dim_customers), on="customer_id")
       .join(broadcast(dim_dates), on="date_id")
   )
   # Каждый join ОЧЕНЬ быстрый!
   

3. Нужна максимальная производительность для одного запроса

   # При наличии памяти, используй broadcast для максимальной скорости
   

❌ НЕ используй broadcast join если:

1. Обе таблицы большие

   # ❌ Ошибка: обе таблицы > 100 MB
   big_table_1 = spark.read.parquet("s3://huge/table1")  # 200 GB
   big_table_2 = spark.read.parquet("s3://huge/table2")  # 150 GB
   
   result = big_table_1.join(
       broadcast(big_table_2),  # НЕПРАВИЛЬНО!
       on="key"
   )
   # Вызовет OutOfMemory на executor
   

2. Памяти недостаточно

   # Если маленькая таблица 5 GB, а памяти на executor 4 GB
   # Broadcast не поместится в памяти
   

3. Маленькую таблицу часто обновляют

   # Если справочник меняется часто,
   # broadcast становится неэффективным
   

Параметры для управления broadcast

# Максимальный размер таблицы для автоматического broadcast
spark.conf.set(
    "spark.sql.autoBroadcastJoinThreshold",
    100 * 1024 * 1024  # 100 MB (по умолчанию)
)

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

# Таймаут для broadcast (сколько времени ждать отправки)
spark.conf.set("spark.sql.broadcastTimeout", "300")  # секунды

# Максимальный размер переменной broadcast
spark.conf.set(
    "spark.broadcast.maxOutBytesInFlight",
    "800m"  # По умолчанию 800 MB
)

Реальные примеры

Пример 1: Обогащение фактов справочниками

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

# Факты
sales = spark.read.parquet("s3://warehouse/sales")

# Справочники (маленькие)
products = spark.read.parquet("s3://dim/products")      # 30 MB
customers = spark.read.parquet("s3://dim/customers")    # 50 MB
stores = spark.read.parquet("s3://dim/stores")          # 5 MB
categories = spark.read.parquet("s3://dim/categories")  # 1 MB

# Объединяем с broadcast
result = (
    sales
    .join(broadcast(products), on="product_id", how="left")
    .join(broadcast(customers), on="customer_id", how="left")
    .join(broadcast(stores), on="store_id", how="left")
    .join(broadcast(categories), on="category_id", how="left")
    .select(
        sales.sale_id,
        products.product_name,
        customers.customer_name,
        stores.store_name,
        categories.category_name,
        sales.amount
    )
)

Пример 2: Поиск значений в справочнике

# Большая таблица сырых данных
raw_data = spark.read.parquet("s3://raw/data")

# Справочник кодов
code_mapping = spark.createDataFrame([
    ("A", "Active"),
    ("I", "Inactive"),
    ("P", "Pending")
], ["code", "status"])

# Добавляем статус
result = raw_data.join(
    broadcast(code_mapping),
    raw_data.status_code == code_mapping.code,
    how="left"
).drop(code_mapping.code)

Пример 3: Filtering с помощью broadcast

# Большая таблица
all_transactions = spark.read.parquet("s3://transactions")

# Маленький справочник с fraud-users
fraud_users = spark.read.parquet("s3://fraud_list")  # 10 MB

# Находим fraudulent transactions
fraud_transactions = (
    all_transactions
    .join(
        broadcast(fraud_users),
        all_transactions.user_id == fraud_users.fraud_user_id,
        how="inner"
    )
)

Планы выполнения

Без broadcast (с shuffle):

BroadcastHashJoin: false
├── Exchange (shuffle)  <- ДОРОГАЯ ОПЕРАЦИЯ
├── Project
└── Filter

С broadcast (без shuffle):

BroadcastHashJoin: true
├── Broadcast (маленькая таблица копируется)
├── Project
└── Filter

Чтобы увидеть план:

result.explain(extended=True)  # Выведет весь план

Типичные проблемы

❌ Проблема 1: OutOfMemory при broadcast больших таблиц

# Неправильно: пытаемся broadcast-ить большую таблицу
big_df = spark.read.parquet("s3://huge/1tb")  # 1 TB
result = df.join(broadcast(big_df), on="key")
# Executor будет пытаться загрузить 1 TB в память -> OOM

Решение: Не broadcast-ь большие таблицы, используй обычный JOIN.

❌ Проблема 2: Timeout при broadcast

# Если маленькая таблица очень большая (100+ MB),
# broadcast может timeout-ться
# Увеличь timeout:
spark.conf.set("spark.sql.broadcastTimeout", "600")  # 10 минут

❌ Проблема 3: Duplicates при broadcast

# Если маленькая таблица имеет дубликаты
# JOIN может создать нежелательные дубликаты

# Решение: удали дубликаты перед broadcast
small_table_unique = small_table.dropDuplicates(["key"])
result = big_table.join(broadcast(small_table_unique), on="key")

Best Practices

1. Всегда проверяй размер таблицы перед broadcast

   # Используй explain() чтобы увидеть какую стратегию выберет Spark
   df.explain()
   

2. Предпочитай broadcast для справочников

   # Справочники обычно маленькие и идеальны для broadcast
   

3. Комбинируй несколько broadcast join-ов

   # Это очень эффективно и масштабируется хорошо
   

4. Убедись, что память достаточна

   # Каждый executor должен иметь достаточно памяти
   # для хранения broadcast таблицы
   

5. Используй явный broadcast() когда уверен

   # Явный broadcast гарантирует стратегию
   result = df1.join(broadcast(df2), on="key")
   

Broadcast join — это мощный инструмент для оптимизации больших аналитических запросов. Правильное использование может ускорить запросы в 10+ раз!