Broadcast Hash Join

Broadcast Hash Join — это алгоритм соединения (join) таблиц в системах управления базами данных (СУБД), используемый в распределённых вычислительных средах, таких как Apache Spark, Hadoop и другие MPP-системы. Он применяется для эффективного выполнения операции соединения, когда одна из таблиц достаточно мала, чтобы её можно было полностью скопировать (разослать) на все узлы кластера, где хранятся фрагменты другой, большой таблицы. Ключевая особенность алгоритма — минимизация перемещения данных по сети за счёт однократной широковещательной рассылки (broadcast) небольшой таблицы и последующего локального хеш-соединения (hash join) на каждом узле.

История и контекст

Алгоритм Broadcast Hash Join является развитием классического хеш-соединения (hash join), адаптированного для параллельных и распределённых архитектур. С ростом популярности распределённых систем обработки данных, таких как MapReduce (середина 2000-х годов) и Apache Spark (начало 2010-х годов), возникла потребность в эффективных способах соединения больших наборов данных без дорогостоящей перетасовки (shuffle) данных по сети. Одним из первых и наиболее успешных решений стал Broadcast Hash Join, который впервые был реализован в Apache Hive (инструмент для работы с данными в Hadoop) и затем широко внедрён в Spark SQL, Presto, Impala и другие системы.

Принцип работы

Broadcast Hash Join состоит из двух основных этапов: широковещательной рассылки (broadcast) и локального хеш-соединения (hash join).

Широковещательная рассылка (Broadcast)

На этом этапе меньшая таблица (обычно называемая «таблицей фактов» или «малой таблицей») полностью копируется на все узлы кластера. В Apache Spark, например, для этого используется метод broadcast(), который помечает таблицу как «широковещательную». Драйвер Spark собирает все данные этой таблицы с узлов, где они хранятся, и отправляет их на каждый исполнитель (executor). Размер передаваемых данных ограничен настройками конфигурации (например, spark.sql.autoBroadcastJoinThreshold, который по умолчанию составляет 10 МБ в Spark 3.x). Если таблица превышает этот порог, Spark переключается на другие алгоритмы (например, Sort Merge Join).

Локальное хеш-соединение (Hash Join)

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

1. Построение хеш-таблицы: На каждом узле из полученной копии малой таблицы строится хеш-таблица по ключу соединения (join key). Хеш-таблица хранится в оперативной памяти узла.
2. Сканирование большой таблицы: Каждый узел сканирует свой фрагмент большой таблицы (которая распределена по узлам). Для каждой строки большой таблицы вычисляется хеш от ключа соединения, и производится поиск в локальной хеш-таблице.
3. Выдача результата: Если совпадение найдено, формируется результирующая строка. Если нет — строка отбрасывается (для inner join) или добавляется с NULL-значениями (для left/right outer join).

Условия применения

Broadcast Hash Join эффективен только при соблюдении ряда условий:

• Размер малой таблицы: Малая таблица должна помещаться в оперативную память каждого узла кластера. В противном случае хеш-таблица будет вытесняться на диск, что резко снизит производительность. Обычно порог устанавливается администратором системы (например, 10–100 МБ).
• Тип соединения: Алгоритм поддерживает большинство типов соединений: inner join, left outer join, right outer join, full outer join, а также semi join и anti join. Однако для full outer join требуется осторожность, так как он может порождать дубликаты при неправильной реализации.
• Равномерное распределение данных: Алгоритм предполагает, что большая таблица равномерно распределена по узлам. Если данные сильно скошены (skewed), некоторые узлы могут получить непропорционально большую нагрузку, что снизит общую производительность.

Преимущества и недостатки

Преимущества

• Минимизация сетевого трафика: В отличие от Sort Merge Join или Shuffle Hash Join, Broadcast Hash Join не требует перетасовки (shuffle) данных большой таблицы. Весь трафик ограничивается однократной рассылкой малой таблицы, что значительно снижает нагрузку на сеть.
• Высокая скорость: При малом размере малой таблицы алгоритм выполняется практически мгновенно, так как все операции — локальные.
• Простота реализации: Алгоритм легко встраивается в существующие системы и не требует сложной координации между узлами.

Недостатки

• Ограничение по размеру: Алгоритм неприменим, если обе таблицы велики. В таких случаях требуется использование более сложных алгоритмов, таких как Sort Merge Join или Shuffle Hash Join.
• Потребление памяти: Каждый узел должен хранить полную копию малой таблицы в оперативной памяти. Если узлов много, это может привести к избыточному расходу памяти.
• Зависимость от пропускной способности сети: Хотя объём передаваемых данных мал, при очень большом количестве узлов (сотни и тысячи) широковещательная рассылка может создать значительную нагрузку на драйвер и сеть.

Примеры использования

Broadcast Hash Join широко применяется в следующих сценариях:

• Соединение справочников с большими таблицами: Например, соединение таблицы «клиенты» (малая, до 1000 записей) с таблицей «продажи» (миллиарды записей) для получения информации о клиентах.
• Фильтрация по измерению: В аналитических запросах (OLAP) часто требуется соединить таблицу фактов с небольшими таблицами измерений (например, «время», «продукт», «магазин»).
• Оптимизация запросов в Spark SQL: Планировщик Spark автоматически выбирает Broadcast Hash Join, если размер малой таблицы меньше заданного порога. Разработчики могут также принудительно указать broadcast в коде.

Реализации в популярных системах

Apache Spark

В Spark SQL Broadcast Hash Join является одним из трёх основных алгоритмов соединения (наряду с Sort Merge Join и Shuffle Hash Join). Планировщик Spark автоматически определяет возможность его применения на основе статистики таблиц (размер, количество строк). Разработчик может явно указать подсказку /+ BROADCAST(t) / в SQL-запросе или использовать broadcast() в Scala/Python API.

Apache Hive

В Hive Broadcast Hash Join реализован в режиме «Map Join». Он выполняется на этапе MapReduce, когда малая таблица загружается в распределённый кэш (DistributedCache) и раздаётся всем mapper'ам. Для этого используется подсказка /+ MAPJOIN(t) /.

Presto и Impala

В Presto и Impala Broadcast Hash Join также является стандартным алгоритмом для соединения малых таблиц с большими. Обе системы автоматически выбирают его, если размер малой таблицы не превышает заданного лимита.

Сравнение с другими алгоритмами

Интересные факты

• В Apache Spark порог автоматического выбора Broadcast Hash Join (spark.sql.autoBroadcastJoinThreshold) по умолчанию составляет 10 МБ. Это значение было выбрано как компромисс между производительностью и риском переполнения памяти.
• В некоторых системах (например, в Spark) Broadcast Hash Join может быть использован для соединения более чем двух таблиц (multi-way join), если одна из них достаточно мала.
• При неправильной настройке (например, слишком большом пороге) Broadcast Hash Join может привести к OutOfMemoryError (ошибке нехватки памяти) на узлах, если малая таблица не помещается в память.

Источники

• Apache Spark Documentation: "Spark SQL, DataFrames and Datasets Guide" (раздел "Join Strategies")
• Apache Hive Language Manual: "MapJoin" (раздел "Optimization")
• Presto Documentation: "Join Optimization" (раздел "Broadcast Join")
• Impala Documentation: "Join Strategies" (раздел "Broadcast Hash Join")
• Статья "Broadcast Hash Join in Apache Spark" на сайте Databricks Engineering Blog
• Книга "Learning Spark" (2-е издание), глава 7 "Optimizing and Tuning Spark SQL"