Оптимизация запросов JOIN в HIVE

В последних версиях HIVE пытается внедрить CBO (cost based optimizer) и JOIN одна из главных его составляющих. Понимание лучших примеров применения соединений является одним из ключевых факторов настройки производительности HIVE.

Рассмотрим каждый вид соединений на примерах и определим их разницу:

Shuffle Join (Common Join) – общее соединение или соединение в случайном порядке

Этот вид соединений используется по умолчанию и включает map и reduce этапы
Mapper: считывает таблицы и выводит пары ключ-значение соединения в промежуточный файл.

Shuffle: эти пары сортируются и объединяются.

Reducer: получает отсортированных данных и делает соединение.

Common Join в Hive
Common Join в Hive

Варианты использования:

Работает для таблиц любого размера, особенно, когда другие тип соединений не могут быть использованы, например, полное внешнее соединение.

Недостатки:

Большая ресурсоемкость, так как shuffle – дорогостоящая операция.

Пример:

select a.* from passwords a, passwords2 b where a.col0=b.col1;

Подсказки:

Самая большая таблица должна быть поставлена ​​справа, так как это должна быть потоковая таблица. 

Однако вы можете использовать подсказку «STREAMTABLE», чтобы изменить таблицу потока на каждом этапе map-reduce.

select /*+ STREAMTABLE(a) */ a.* from passwords a, passwords2 b, passwords3 c where a.col0=b.col0 and b.col0=c.col0;

 

Map Join (Broardcast Join)

Если одна или несколько таблиц достаточно малы, чтобы поместиться в памяти, mapper просматривает большую таблицу и делает соединение. Отсутствует этап  перемешивания (shuffle) и свертки (reduce).

Использование Hash Join в Hive
Использование Hash Join в Hive

Варианты использования:

Существует маленькая, которую необходимо присоединить к большой.

Недостатки:

Требуется чтобы по крайней мере одна таблица была достаточно мала.

Правое/полное внешнее соединение не работает.

Пример:

Здесь очень маленькой таблицей является — passwords, большой – passwords2.

select /*+ MAPJOIN(a) */ a.* from passwords a, passwords2 b where a.col0=b.col0 ;

Данная подсказка используется если параметр hive.ignore.mapjoin.hint = false (по умолчанию true в Hive 0.13)

Подсказки:

Автоматическое преобразование общего соединения common join в map join

Необходимо  настроить следующие конфигурационные параметры в hive-site.xml или непосредственно из оболочки Hive.

set hive.auto.convert.join=true;

set hive.auto.convert.join.noconditionaltask=true;

А также параметр set hive.auto.convert.join.noconditionaltask.size в зависимости от размера данных

Bucket Map Join – соединение сегментами

Соединение выполняется только в mapper. Mapper обработки сегмента 1 для таблицы A будет получать только сегмент 1 из таблицы B.

Особенности Bucket Join в Hive
Особенности Bucket Join в Hive

Варианты использования:

Когда все таблицы большие b количество сегментов в одной таблице кратно количеству сегментов в другой таблице.

Недостатки:

Таблицы должны быть сегментированы по тем же полям, которые используются для соединения.

Пример:

create table b1(col0 string,col1 string,col2 string,col3 string,col4 string,col5 string,col6 string)

clustered by (col0) into 32 buckets;

create table b2(col0 string,col1 string,col2 string,col3 string,col4 string,col5 string,col6 string)

clustered by (col0) into 8 buckets;

set hive.enforce.bucketing = true;

from passwords insert OVERWRITE  table b1 select * limit 10000;

from passwords insert OVERWRITE  table b2 select * limit 10000;

set hive.optimize.bucketmapjoin=true;

select /*+ MAPJOIN(b2) */ b1.* from b1,b2 where b1.col0=b2.col0 ;

Skew Join

Такой вид соединений используется если таблица A соединяется с таблицей B, но A содержит перекошенные данные (т.е. содержит значения, которые появляются очень часто) в столбце соединения.

Необходимо:

  • Прочитать таблицу B и сохранить строки с ключом в хэш-таблице в памяти
  • Запустить mapper для чтения A и выполнить следующие действия:
    • если ключ совпадает использовать хэшированную версию B  для вычисления результата
    • остальные ключи отправляются на reducer , который делает соединение. Этот reducer будет получать строки из B также из mapper.

Таким образом, мы считываем B дважды, перекошенные ключи в A прочитаны и обработаны только mapper и не отправляются к reducer . Остальные ключи в A проходят через один этап map-reduce.

Особенности Skew Join в Hive
Особенности Skew Join в Hive

Варианты использования:

Одна таблица должна иметь перекошенные значения в столбце соединения.

Недостатки:

Одна таблица читается дважды.

Перекос в данных должно быть известно.

Пример:

Должен быть установлен параметр для включения соединения с перекосом:

set hive.optimize.skewjoin = true;

И параметр с количеством строк, который показывает, что ключ является перекошенным, если имеет значение больше указанного

set hive.skewjoin.key=100000;

select a.* from passwords a, passwords2 b where a.col0=b.col1;

Подсказки:

Начиная с Hive 0.10.0, таблицы могут быть созданы как перекошенные. Указав значения, которые появляются очень часто, Hive автоматически разбивает их на отдельные файлы и учитывает этот факт во время запросов, чтобы он мог пропустить или включить весь файл, если это возможно.

CREATE TABLE list_bucket_single (key STRING, value STRING)

 SKEWED BY (key) ON (1,5,6) [STORED AS DIRECTORIES];

В этом примере показан один столбец с тремя перекошенными значениями, дополнительно используется параметр STORED AS DIRECTORIES, который определяет сегментирование списка.

Источник: http://www.openkb.info/2014/11/understanding-hive-joins-in-explain.html