Аналитика больших данных с Apache Hive: оптимизация SQL-запросов

Apache Hive – востребованный инструмент класса SQL-on-Hadoop, который также активно используется в работе с фреймворком Spark. Поэтому сегодня разберем важную тему из обучения дата-инженеров и аналитиков больших данных про оптимизацию SQL-запросов в этом NoSQL-хранилище. Смотрите, чем полезна векторизация HiveQL-операций, какие форматы файлов обрабатываются быстрее, почему денормализация данных в Hive – это хорошо, в чем разница между Spark и Tez, зачем выбирать движок исполнения вместо MapReduce, как посмотреть план выполнения запроса и включить CBO-оптимизацию.

5 методов оптимизации SQL-запросов в Apache Hive

Напомним, Apache Hive — это аналитический инструмент к хранилищу данных на Hadoop HDFS с помощью SQL-подобного языка структурированных запросов HiveQL (HQL). Не являясь реляционной СУБД, Hive не слишком подходит для анализа и транзакционных обновлений в реальном времени. Основными компонентами Apache Hive как средства SQL-on-Hadoop являются следующие [1]:

пользовательские интерфейсы веб-GUI и CLI, которые обеспечивают взаимодействие между пользователем и распределенной файловой системой Hadoop (HDFS, Hadoop Distributed File System);
хранилище метаданных (Meta Store), где хранятся схемы таблиц, баз данных и сопоставления с HDFS;
механизм обработки HQL-запросов как абстракция заданий MapReduce;
механизм выполнения запросов и получения результатов, использующий задания MapReduce;
HDFS, где лежат табличные данные Hive.

С учетом архитектурных особенностей Apache Hive, о которых мы писали здесь, для нее характерны довольно высокие задержки в обработке данных и пакетный характер вычислений. Поэтому ускорение SQL-запросов с помощью их оптимизации весьма актуально для этого NoSQL-хранилища. Основными методами оптимизации SQL-запросов в Apache Hive являются следующие [1]:

партиционирование или/и бакетирование таблиц;
хранение файлов в определенных форматах, сжатых специальным образом;
оптимизация SQL-операций;
выбор наиболее быстрого механизм исполнения;
векторизация.

Каждый из этих методов мы подробнее рассмотрим дальше.

Оптимизация на уровне Hive-таблицы: партиционирование, бакетирование и денормализация

Оптимизация на уровне структуры таблицы помогает улучшить управление данными. Таблицы Hive поддерживают 2 следующих способа оптимизации, которые могут использоваться одновременно:

партиционирование – логическое разделение таблицы на подмножества на основе определенных значений столбцов, чтобы сократить объем сканирования при поиске данных. Подходит для столбцов с низкой мощностью в качестве ключа партиционирования. Например, если данные связаны с измерением времени, то дата – подходящий ключ раздела. Или для данных, связанных с местоположением (страна, область, город и пр.), подойдут иерархические разделы, например, страна/область.
бакетирование — разделение данных на подмножества файлов по хэш-функции столбца, чтобы ускорить последовательные чтения данных для последующих заданий. Подходит для столбцов с высокой мощностью, о чем мы подробно рассказывали здесь на примере Spark.

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

динамическое, которое система выполняет сама. Это достаточно медленно и по умолчанию отключено. Строгий режим (sirict) динамического партиционирования требует хотя бы одного столбца со статическим разделом. столбец статического раздела. Нестрогий режим (non strict) позволяет иметь динамические значения на всех разделах. Перед установкой динамического партиционирования следует задать эти свойства:

set hive.exec.dynamic.partition=true;

set hive.exec.dynamic.partition.mode=nonstrict;

Дополнительно можно настроить число Reducer’ов, выполняющих 2-ой шаг вычислительной модели MapReduce, в которую транслируется SQL-запроc Hive:

hive.exec.reducers.bytes.per.reducer;

hive.exec.reducers.max;

mapreduce.job.reduces.

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

Бакетирование таблиц в Apache Hive похоже на партиционирование, но данные разделяются сразу на файлы вместо каталогов, что подходит для большого количества столбцов разделения, т.е. множества различных значений. Таблицы или разделы делятся на под-файлы на основе хэш-функции столбца и группируются в один файл, чтобы сканировать только нужные файлы. По умолчанию бакетирование отключено, для включения нужно установить свойство hive.enforce.bucketing в значение true [1].

Также для этого NoSQL-хранилища рекомендуется денормализация данных, чтобы сократить число операций соединения таблиц, которые всегда являются дорогостоящими и вызывают проблемы с производительностью [2].

Форматы файлов и кодеки сжатия

Поскольку Hive хранит данные в виде файлов в Hadoop HDFS, то знание форматов, которые поддерживает эта файловая система поможет оптимизировать работу по их обработке. По умолчанию Hive использует формат текстового файла, но также поддерживает следующие популярные форматы:

В частности, ORC может уменьшить объем хранилища данных на 75%, используя методы predicate push-down и сжатия для повышения производительности запроса. Например, кодекс Snappy обеспечивает быстрое сжатие. Колоночно-ориентированная структура бинарного формата Parquet также намного повышает скорость чтения данных за счет работы с отдельными столбцами вместо считывания всего файла. Таким образом, колоночные форматы позволяют сократить операции чтения в аналитических запросах, разрешая доступ к каждому столбцу индивидуально [3].

Сжатие уменьшает объем промежуточных данных, что сводит к минимуму объем передачи информации по сети между Map и Reduce. Сжатие может применяться к выходу Mapper’а и Reducer’а по отдельности. Но файлы, сжатые с помощью gzip, нельзя разделить, поэтому данный метод следует применять с осторожностью. Кроме того, размер сжатого файла должен быть не более пары сотен мегабайт, чтобы не привести к дисбалансу задания. Альтернативами gzip-сжатия являются кодеки Snappy, lzo, bzip и пр. [2]

Оптимизация SQL-запроса на уровне операций

Чтобы ускорить выполнение запроса в распределенных системах, нужно свести перемешивание к минимуму. В случае операций соединения (JOIN) для этого используются Map side joins, Bucket Map Join, Sort Merge Bucket Join [3].

Map-соединения эффективны, если одна из таблиц может уместиться в памяти. За их автоматическое выполнение отвечает параметр hive.auto.convert.join, установленный в значение true. При этом следует также включать автоматическое преобразование и бакетирование (SET hive.enforce.bucketing = true;) каждый раз перед записью данных в бакетированную таблицу.

Для операции соединения нужно еще включить параметр hive.optimize.bucketmapjoin, который указывает Hive, что JOIN будет выполняться на уровне бакета во время этапа Map. Это сокращает циклы сканирования для поиска определенного ключа, т.к. бакетирование гарантирует, что ключ присутствует в определенном бакете [2].

Пользовательские функции (UDF) обычно не очень хорошо оптимизируются, а операции фильтрации оцениваются слева направо. Поэтому для лучшей производительности рекомендуется поместить UDF справа в список выражений с оператором AND в условии WHERE.

Следует избегать связанных запросов и встроенных таблиц. Лучше создавать временные таблицы и использовать внутреннее соединение вместо внешнего. Альтернативой временным таблицам является выражение WITH. Также следует избегать любых SQL-запросов, которые приводят к декартову соединению (CARTESIAN JOIN) [4].

Посмотрев план выполнения SQL-запроса, сгенерированный Hive по принципу оптимизации затрат (Cost Based Optimization, CBO), можно сократить время выполнения и уменьшить использование ресурсов. Сделать это поможет оператор EXPLAIN, добавленный перед запросом. А команда ANALYZE TABLE [tbl_name] COMPUTE STATISTICS соберет по таблице следующую статистику [3]:

Количество строк;
Количество файлов;
Размер в байтах;
Номер раздела для партиционированной таблицы.

Чтобы использовать CBO-оптимизацию, нужно включить следующие параметры [4]:

set hive.cbo.enable=true;
set hive.compute.query.using.stats=true;
set hive.stats.fetch.column.stats=true;
set hive.stats.fetch.partition.stats=true.

Подробнее о том, как устроен язык запросов HiveQL, мы писали здесь.

Механизм исполнения и параллелизм

Hive поддерживает различные механизмы выполнения SQL-запросов: классический MapReduce, Tez и Spark. В большинстве случаев рекомендуется избегать использования MapReduce, отдавая предпочтение Tez или Spark, установив это через конфигурацию hive.execution.engine, например, set hive.execution.engine = spark. Tez и Spark поддерживают парадигму DAG (Directed Acyclic Graph), позволяя в рамках одного DataFlow параллельно выполнять несколько разных процессов (Mapper или Reducer) с готовыми данными.

Выбирая между Spark и Tez, стоит помнить, что Tez прекрасно вписывается в архитектуру YARN, а Spark может столкнуться с проблемами управления ресурсами. Однако, универсальный Spark отлично подходит большинства задач, а Tez всего лишь фреймворк для специализированных инструментов. Контейнеры Tez могут отключаться по завершении задания для экономии ресурсов, а контейнеры Spark потребляют ресурсы, даже если они не обрабатывают данные. Начиная с версии 0.13, Hive использует Tez в качестве основного механизма выполнения, что приводит к значительному повышению производительности [5].

Однако, быстрый Tez менее устойчив к неравномерному распределению данных, чем MapReduce, требует тонкой настройки конфигурационных параметров и со временем порождает слишком много классов, что чревато переполнением Garbage Collector, вызывающим перезагрузку с потерей временных таблиц [6].

Поэтому следует очень аккуратно выбирать между движком выполнения SQL-запросов в Apache Hive, принимая во внимание, что классический Hadoop MapReduce тоже может распараллеливать задания, правда, намного примитивнее, чем DAG-механизмы. Некоторые сложные запросы Hive обычно транслируются в несколько заданий MapReduce, которые выполняются по умолчанию последовательно. Но отдельные этапы MapReduce запроса могут выполняться параллельно, повышая эффективность утилизации ресурсов Hive-кластера и снижая общее время выполнения SQL-запроса. Для этого в следует установить конфигурацию hive.exec.parallel в значение true [2]. Подробнее о сходстве и отличиях Apache Spark и Tez читайте в нашей новой статье.

Векторизация HiveQL-операций

Векторизация повышает производительность операций HiveQL, таких как сканирование, агрегирование, фильтры и объединения, за счет их одновременного выполнения пакетами по 1024 строки вместо одной строки за раз. Каждый пакет состоит из вектора-столбца, который обычно представляет собой массив примитивных типов. Операции выполняются над всем вектором-столбцом, что улучшает конвейеры команд и использование кэша. Чтобы включить векторизацию, установите следующие параметры конфигурации [4]:

set hive.vectorized.execution.enabled = true;
set hive.vectorized.execution.reduce.enabled = true.

Узнайте больше про администрирование и эксплуатацию Apache Hive для разработки распределенных приложений и аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:

Смотреть расписание

Записаться на курс

Источники