Как работают hash join и merge join
Что проверяют
Понимаете ли вы, что происходит при соединении двух больших таблиц в Spark и почему один и тот же join идёт то минуту, то час. Следом обычно спрашивают, что делать, если одна сторона маленькая, и что такое перекос ключей.
Как отвечать
Сначала база, общая для СУБД и Spark:
- hash join: по одной стороне строится хеш-таблица, вторая ищет в ней совпадения. Только равенство, быстро, но таблица должна поместиться в память;
- sort-merge join: обе стороны сортируются по ключу и сливаются двумя курсорами. Дороже из-за сортировки, зато сортировка умеет уходить на диск, поэтому метод надёжно масштабируется.
Дальше распределённый случай в Spark:
- broadcast hash join: маленькая таблица целиком копируется на каждый executor, и большая таблица не перемешивается по сети. Spark выбирает его сам, если сторона меньше
spark.sql.autoBroadcastJoinThreshold(по умолчанию 10 МБ), или по подсказкеbroadcast(df); - sort-merge join: выбор по умолчанию для двух больших таблиц. Обе стороны перемешиваются (shuffle) по ключу в
spark.sql.shuffle.partitionsпартиций, по умолчанию 200, сортируются и сливаются; - shuffle hash join: перемешивание без сортировки, хеш-таблица строится в каждой партиции. Выгоден, когда одна сторона заметно меньше, но в broadcast не влезает; включается подсказкой
SHUFFLE_HASH.
Adaptive Query Execution (по умолчанию с Spark 3.2) смотрит на реальные размеры после shuffle, может превратить sort-merge в broadcast и делит перекошенные партиции. Выбор виден в df.explain() и во вкладке SQL в Spark UI: BroadcastHashJoin, SortMergeJoin и узлы Exchange.
Где спотыкаются
Ставят broadcast на таблицу в несколько гигабайт. Драйвер собирает её целиком, и задача падает по памяти раньше, чем начнётся join.
Не проверяют перекос. Один частый ключ, например NULL или «unknown», собирает миллионы строк в одну партицию, и вся стадия ждёт одну задачу. Помогают фильтрация пустых ключей до join, соль в ключе и skew join в AQE.
Считают sort-merge медленным и пытаются везде навязать hash. Для двух больших таблиц он надёжнее: хеш-таблице нужна память, а сортировка переживает её нехватку через сброс на диск.