Обзор

Apache Hive – система управления (СУБД) реляционными базами данных (РБД) с открытым исходным кодом для запросов, агрегирования и анализа параметров и режимов рабочих нагрузок с большими данными. В этой статье описываются ключевые инновационные инструменты для полноценной пакетной обработки в корпоративной системе хранения данных. Мы представляем гибридную архитектуру, которая сочетает в себе традиционные методы массивно-параллельных архитектур (MPP) с физически разделенной памятью с более современными концепциями больших данных, облаков для достижения масштабируемости и производительности, требуемых современными аналитическими приложениями. Мы исследуем систему, подробно описывая улучшения по четырем основным направлениям: транзакция, оптимизатор, среда выполнения и федерация (интеграционный процесс). Затем мы приводим экспериментальные результаты, чтобы продемонстрировать производительность системы для типовых рабочих нагрузок, и в заключение рассмотрим дорожную карту сообщества.

1. ВВЕДЕНИЕ

Когда Hive был впервые представлен более 10 лет назад [55], мотивацией авторов было предоставить SQL-подобный интерфейс поверх Hadoop MapReduce, чтобы освободить пользователей от работы с деталями реализации низкого уровня и сосредоточить их на задания параллельной пакетной обработки. Hive сосредоточился в основном на рабочих нагрузках ExtractTransform-Load (ETL) или на пакетных отчетах, которые состояли из:

чтения огромных объемов данных;
выполнения преобразований над этими данными (например, разбор данных, консолидация, агрегация);
загрузки выходных данных в другие системы, которые использовались для дальнейшего анализа.

Поскольку Hadoop стала массовой платформой для недорогого хранения данных с HDFS, разработчики сосредоточились на увеличении диапазона рабочих нагрузок, которые могли бы эффективно выполняться в рамках платформы. Была введена YARN [56], структура управления ресурсами для Hadoop, а сразу после этого, были включены для работы на Hadoop непосредственно с помощью поддержки YARN и механизмы обработки данных (кроме MapReduce), в частности, Spark [14, 59], Flink [5, 26].

Пользователи также все больше внимания уделяют переносу рабочих нагрузок хранилищ данных из других систем в Hadoop. Эти касается таких нагрузок, как интерактивные и специальные отчеты, информационные панели и другие варианты использования бизнес-аналитики. Существовало общее требование, которое затрудняло анализ использования рабочих нагрузок в Hadoop: они требовали SQL-движка с низкой задержкой. Параллельно предпринимались многочисленные усилия по достижению этой цели, появились новые системы SQL MPP, совместимые с YARN, такие как Impala [8, 42] и Presto [48].

Вместо того, чтобы внедрять новую систему, сообщество Hive пришло к выводу, что нынешняя реализация проекта обеспечивает хорошую основу для поддержки этих рабочих нагрузок. Hive был разработан для крупномасштабных надежных вычислений в Hadoop и он уже обеспечивал совместимость с SQL (увы, ограниченную) и подключение к другим СУБД. Однако Hive необходимо было эволюционировать и, самоорганизуясь, подвергнуться капитальному обновлению, чтобы удовлетворять требованиям новых вариантов использования. Для этого – принять общие методы хранения данных, которые были тщательно изучены на протяжении многих лет.

Предыдущие работы по Hive, представленные исследовательскому сообществу, были сосредоточены на его:

первоначальной архитектуре и реализации поверх HDFS и MapReduce [55];
улучшениях для устранения многочисленных недостатков производительности в исходной системе, включая введение оптимизированного формата файлов по столбцам, физическую оптимизацию для уменьшения количества фаз MapReduce в планах запросов и векторизованную модель реализации для повышения эффективности выполнения [39].

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

Поддержка SQL и ACID (Раздел 3). Соответствие требованиям SQL является ключевым требованием для хранилищ данных. Таким образом, поддержка SQL в Hive была расширена, включая, среди прочего, коррелированные подзапросы, ограничения целостности и расширенные операции OLAP. В свою очередь, пользователи хранилища нуждаются в поддержке вставок, обновлений, удалений и объединений своих отдельных записей «как есть» («as for as»). Hive предоставляет гарантии ACID с изоляцией моментальных снимков файловой системы виртуальной машины (для «отката» в работоспособное состояние) с помощью диспетчера транзакций, построенного поверх Метастора для хранения схемы и статистики по которому Hive определяет, как получить доступ к данным.

Методы оптимизации (Раздел 4). Оптимизация запросов особенно актуальна для систем управления данными, использующих декларативные языки запросов, такие как SQL. Вместо того, чтобы внедрять свой собственный оптимизатор с нуля, Hive решает интегрироваться с Calcite [3, 19] и привнести свои возможности оптимизации в систему. Кроме того, Hive включает в себя другие методы оптимизации, обычно используемые в средах хранилищ данных, такие как повторная оптимизация запросов, кэширование результатов запросов и перезапись материализованных представлений.

Задержка выполнения (Раздел 5). Чтобы охватить более широкий спектр вариантов использования, включая интерактивные запросы, крайне важно улучшить задержку. Поддержка Hive для оптимизированного хранения вектор-столбцов данных и векторизации операторов была представлена в предыдущей работе [39]. С тех пор, в дополнение к этим улучшениям, Hive перешел с MapReduce на систему Tez [15, 50], совместимую с YARN среду выполнения, которая обеспечивает большую, чем MapReduce, гибкость при реализации произвольных приложений обработки данных. Кроме того, Hive включает в себя LLAP, дополнительный уровень постоянных длительных исполнителей, который обеспечивает кэширование данных, оптимизацию среды выполнения и позволяет избежать издержек при распределении контейнеров YARN при запуске.

Возможности федерации (Раздел 6). Одной из наиболее важных особенностей Hive является способность обеспечить единый уровень SQL поверх многих специализированных СУБД, появившихся за последние несколько лет. Благодаря интеграции с Calcite и улучшениям обработчика хранения, Hive может легко выполнять вычисления и считывать данные из этих систем. В свою очередь, реализация легко расширяется и адаптируется для поддержки других систем в будущем.

Остальная часть статьи организована следующим образом. В разделе 2 представлена справочная информация об архитектуре и основных компонентах Hive. В разделах 3-6 описан наш основной вклад в совершенствование системы по указанным выше четырем направлениям. В разделе 7 представлена экспериментальная оценка Hive. В разделе 8 обсуждается влияние новых функций, а в разделе 9 кратко описывается дорожная карта проекта. Наконец, в разделе 10 кратко излагаются наши выводы.

2. АРХИТЕКТУРА СИСТЕМЫ

В этом разделе мы кратко познакомимся с архитектурой Hive. На рис. 1 показаны основные компоненты системы.

Хранение данных. Данные в Hive могут храниться в любом из поддерживаемых форматов файлов в любой файловой системе, совместимой с Hadoop. На сегодняшний день наиболее распространенными форматами файлов являются ORC [10] и Parquet [11]. В свою очередь, совместимые файловые системы включают HDFS, которая является наиболее часто используемой реализацией распределенной файловой системы для всех основных коммерческих хранилищ облачных объектов, таких как AWS S3 и Blob–объектов Azure. Кроме того, Hive также может считывать и записывать данные в другие автономно используемые системы обработки, например, Druid [4, 58] или HBase [6], которые мы более подробно обсудим в разделе 6.

Каталог данных. Hive хранит всю информацию о своих источниках данных с помощью метастора Hive (или HiveMetaStore, проще говоря). В двух словах, HMS-это каталог для всех данных, запрашиваемых Hive. Он использует СУБД для сохранения информации и полагается на DataNucleus [30], реализацию объектно-реляционного отображения Java, чтобы упростить поддержку нескольких СУБД на серверной части. Для вызовов, требующих низкой задержки, HMS может обойти DataNucleus и напрямую запросить RDMBS. API HMS поддерживает несколько языков программирования, и служба реализована с использованием Thrift [16], программной платформы, которая обеспечивает язык определения интерфейса, механизм генерации кода и реализацию двоичного протокола связи.

Сменная среда обработки данных. Hive стал одним из самых популярных SQL-движков от Hadoop, он постепенно отошел от MapReduce, чтобы поддерживать более гибкое время выполнения обработки, совместимое с YARN [50]. Хотя MapReduce все еще поддерживается, в настоящее время наиболее популярной средой выполнения для Hive является Tez [15, 50]. Это среда, которая обеспечивает большую гибкость, чем MapReduce, моделируя обработку данных в виде DAG с вершинами, отображающими логику приложения, и ребрами, представляющими передачу данных, подобно системе Dryad [40] или Hyracks [22]. Кроме того, Tez совместим с LLAP, уровнем постоянного выполнения и кэша, представленным в разделе 5.

Рисунок 2. Этапы подготовки запроса в HiveServer2

(

На рис. 2 показаны этапы, через которые проходит SQL-запрос в HS2, чтобы стать исполняемым по плану. Как только пользователь отправляет запрос в HS2, запрос обрабатывается драйвером, который анализирует оператор и генерирует логический план Calcite [3, 19] изнутри его AST. Затем план Calcite оптимизируется. Обратите внимание, что HS2 получает доступ к информации об источниках данных в HMS для целей проверки и оптимизации. Затем план преобразуется в физический план, потенциальным введением дополнительных операторов для разделения данных, сортировки и т.д. HS2 выполняет дополнительные оптимизации в DAG физического плана, и если поддерживаются все операторы и выражения в плане, на его основе может быть сгенерирован векторизованный план [39]. Физический план передается компилятору задач, который разбивает дерево операторов на группу DAG-исполняемых задач. Hive реализует отдельный компилятор задач для каждой поддерживаемой среды выполнения обработки (Tez, Spark и MapReduce). После создания задач, драйвер отправляет их диспетчеру приложений среды для выполнения в YARN, который обрабатывает выполнение. Для каждой задачи сначала инициализируются физические операторы в рамках этой задачи, а затем они обрабатывают входные данные конвейерным способом. После завершения выполнения, драйвер извлекает результаты запроса и возвращает их пользователю.

3. ПОДДЕРЖКА SQL И ACID

Стандартные транзакции SQL и ACID являются критическими требованиями в корпоративных хранилищах данных. В этом разделе мы представляем расширенную поддержку SQL Hive. Кроме того, мы описываем улучшения, внесенные в Hive, чтобы обеспечить гарантии ACID поверх Hadoop.

3.1 Поддержка SQL

Чтобы обеспечить замену традиционным хранилищам данных, Hive расширили, чтобы поддерживать больше функций ядра SQL. Hive использует внутреннюю модель данных, поддерживающую все основные атомарные типы данных SQL, а также неатомарные типы, такие как STRUCT, ARRAY и MAP. Кроме того, каждый новый выпуск Hive увеличивал поддержку важных конструкций, которые являются частью спецификации SQL. Например, существует расширенная поддержка коррелированных подзапросов, ссылающихся на столбцы из внешнего запроса, расширенных операций OLAP, таких, как наборы группировок или интерфейсные функции, операции набора и ограничения целостности. С другой стороны, Hive сохранил множество функций исходного языка запросов, которые были ценны для его пользовательской базы. Одной из наиболее популярных функций является возможность указать физическую компоновку хранилища во время создания таблицы с помощью предложения PARTITIONED BY <COLUMNS>. В двух словах, предложение позволяет пользователю разделять таблицу по горизонтали. Затем Hive хранит данные для каждого набора значений разделов в другом каталоге файловой системы. Чтобы проиллюстрировать эту идею, рассмотрим следующее определение таблицы и соответствующий физический макет, изображенный на рис. 3:

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

 CREATE TABLE store_sales (
 sold_date_sk INT, item_sk INT, customer_sk INT, store_sk INT,
 quantity INT, list_price DECIMAL(7,2), sales_price DECIMAL(7,2)
 ) PARTITIONED BY (sold_date_sk INT);

3.2 Внедрение ACID

Изначально Hive поддерживал только вставку и удаление полных разделов из таблицы [55]. Хотя отсутствие операций на уровне строк было приемлемым для рабочих нагрузок ETL, по мере развития Hive и для поддержки многих традиционных рабочих нагрузок хранилищ данных возросли требования к полной поддержке DML и транзакциям ACID. Поэтому Hive теперь включает поддержку выполнения инструкций INSERT, UPDATE, DELETE и MERGE. Он обеспечивает гарантии ACID с помощью изоляции моментальных снимков [24], что является гарантией того, что на практике будут считаны последние зафиксированные значения, которые существовали на момент ее запуска, а сама операция успешно зафиксируется лишь, если обновления не конфликтуют с какими-то параллельными обновлениями, сделанными после этого снимка. Это важно для чтения и четко определенной семантики в случае сбоя с использованием встроенного в систему управления транзакциями. В настоящее время транзакции могут охватывать только один оператор; мы планируем поддерживать транзакции с несколькими операторами в ближайшем будущем. Однако можно выполнять запись в несколько таблиц в рамках одной транзакции с помощью операторов Hive с несколькими вставками [55]. Основные проблемы, которые необходимо преодолеть при поддержке операций на уровне строк в Hive:

отсутствие менеджера транзакций в системе;
отсутствие поддержки обновлений файлов в базовой файловой системе.

Ниже мы приводим более подробную информацию о внедрении ACID в Hive и о том, как были решены эти проблемы.

Управление транзакциями и блокировками. HIVE хранит информацию о транзакциях и состоянии блокировки в HMS. Он использует глобальный идентификатор транзакции или TxnId, т.е. монотонно увеличивающуюся величину, генерируемую Метастором, для каждой транзакции, выполняемой в системе. В свою очередь, каждое значение TxnId сопоставляется с одним или несколькими идентификаторами записи или идентификаторами записи. Идентификатор записи – это монотонно увеличивающаяся величина, также генерируемая Метастором, но в пределах области задания таблицы. Идентификатор записи хранится с каждой записью по транзакции; все записи, записанные одной и той же транзакцией в одну и ту же таблицу, имеют один и тот же идентификатор записи. В свою очередь, файлы с одинаковым идентификатором записи идентифицируются уникальным образом с помощью идентификатора файла, в то время как каждая запись в файле идентифицируется уникальным образом с помощью поля RowId. Обратите внимание, что комбинация WriteId, FileId и RowId однозначно идентифицирует каждую запись в таблице. Операция удаления в Hive реализуется как вставка помеченной записи, которая указывает на уникальный идентификатор удаляемой записи.

Для достижения изоляции моментальных снимков HS2 получает логический моментальный снимок данных, которые необходимо прочитать при выполнении запроса. Снимок представлен списком транзакций, содержащим самый высокий ранг TxnId на данный момент (High Water Mark) и набор открытых и прерванных транзакций под ним. Для каждой таблицы, которую должен прочитать запрос, HS2 сначала генерирует список WriteId из списка транзакций, связавшись с HMS; список WriteId аналогичен списку транзакций, но в пределах одной таблицы. Каждая операция сканирования в плане привязана к списку WriteId во время компиляции. Считыватели в этом сканировании будут пропускать строки, идентификатор записи которых выше верхнего водяного знака или является частью набора открытых и прерванных транзакций. Причина сохранения, как глобальных идентификаторов, так и идентификаторов для каждой таблицы заключается в том, что считыватели для каждой таблицы сохраняют меньшее состояние, что становится критичным для производительности при наличии большого количества открытых транзакций в системе.

Степень детализации блокировки является критерием для разделения (секционирования) таблиц, в то время как полная таблица должна быть заблокирована для неразделенных таблиц. HS2 требуется только для получения эксклюзивных блокировок по операциям, которые нарушают работу при чтении и записи, таких как удаление РАЗДЕЛОВ или инструкций DROP TABLE. Все остальные распространенные операции просто получают общие блокировки. Обновления и удаления используют оптимистичное разрешение конфликтов, отслеживая свои наборы записей и разрешая конфликт во время фиксации (позволяя первой фиксации выиграть).

Расположение данных и файлов. Hive хранит данные для каждой таблицы и раздела в отдельном каталоге (вспомните рисунок 3). Подобно [45], мы используем разные хранилища или каталоги в каждой таблице или каждом разделе для поддержки одновременных операций чтения и записи: base и delta, которые, в свою очередь, могут работать с одним или несколькими файлами. Файлы в базовом хранилище содержат все допустимые записи до определенного идентификатора записи. Например, папка base_100 содержит все записи до writeid 100. С другой стороны, дельта-каталог содержит файлы с записями в диапазоне WriteId. Hive хранит отдельные дельта-каталоги для вставленных или удаленных записей (операции обновления разделены на операции удаления и вставки). Операция вставки или удаления создает дельта-каталог с записями, привязанными к одному идентификатору записи, например, delta_101_101 или delete_delta_102_102. Дельта-каталоги, содержащие более одного идентификатора записи, создаются как часть процесса сжатия (описано ниже).

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

Уплотнение. Сжатие – процесс в Hive, который объединяет файлы в каталогах delta с другими файлами в каталогах delta (результат называется незначительным уплотнением) или файлов в каталогах delta с файлами в базовых каталогах (результат называется основным уплотнением). Важнейшими причинами периодического запуска сжатия являются:

уменьшение количества каталогов и файлов в таблицах, которые без сжатия могут повлиять на производительность файловой системы;
сокращение усилий считывания по объединению файлов во время выполнения запроса;
сокращение набора открытых и прерванных идентификаторов TXNID и идентификаторов записи, связанных с каждым снимком, т.е. основное сжатие удаляет историю, увеличивая идентификатор TxnId, при котором все записи в таблицах, как известно, являются действительными.

Сжатие автоматически запускается HS2 при превышении определенных пороговых значений, например, по количеству дельта-файлов в таблице или соотношению записей в дельта-файлах к базовым файлам. Для уплотнения не требуется никаких блокировок над таблицей. Фактически, фаза очистки разделена для фазы объединения, чтобы любой текущий запрос мог завершить свое выполнение до удаления файлов из системы.

4. ОПТИМИЗАЦИЯ ЗАПРОСОВ

Хотя поддержка оптимизации в начальных версиях Hive была ограниченной, очевидно, что разработка внутренних компонентов ее выполнения недостаточна для обеспечения эффективной производительности. Таким образом, в настоящее время проект включает в себя многие сложные методы, обычно используемые в системах РБД. В этом разделе описываются наиболее важные функции оптимизации, которые помогают системе создавать лучшие планы и улучшать выполнение запросов, включая ее фундаментальную интеграцию с Apache Calcite.

4.1 Оптимизатор на основе правил и затрат

Первоначально Apache Hive выполнил несколько перезаписей для повышения производительности при анализе входного оператора SQL. Кроме того, он содержал оптимизатор на основе правил, который применял простые преобразования к физическому плану, сгенерированному на основе запроса. Например, целью многих из этих оптимизаций была попытка минимизировать затраты на перетасовку данных, критическую операцию в движке MapReduce. Были и другие оптимизации для удаления предикатов фильтрации, проецирования неиспользуемых столбцов и удаления разделов. Хотя это было эффективно для некоторых запросов, работа с представлением физического плана делала реализацию сложных перезаписей, таких как изменение порядка соединений, упрощение и распространение предикатов или перезапись на основе материализованного представления, чрезмерно сложной.

По этой причине было введены новое представление плана и оптимизатор на базе Apache Calcite [3, 19]. Calcite – это модульный и расширяемый оптимизатор запросов со встроенными элементами, которые можно комбинировать различными способами для создания собственной логики оптимизации. К ним относятся различные правила переписывания, планировщики и модели затрат.

Calcite предоставляет два разных механизма планирования:

планировщик на основе затрат – запуска правила перезаписи с целью снижения общей стоимости выражения;
исчерпывающий планировщик – запуска правила, пока не создаст выражение, которое больше не изменяется никакими правилами.

Правила преобразования работают нечетко с обоими планировщиками.

Hive реализует многоступенчатую оптимизацию, аналогичную другим оптимизаторам запросов [52], где на каждом этапе оптимизации используется планировщик и набор правил перезаписи. Это позволяет Hive сократить общее время оптимизации, направляя поиск по различным планам запросов. Некоторые из правил кальцита, включенных в Apache Hive, включают изменение порядка соединений, изменение порядка и устранение нескольких операторов, постоянное сворачивание и распространение, а также преобразования на основе ограничений.

Статистика. Статистические данные таблицы хранятся в HMS и предоставляются Calcite во время планирования. К ним относятся количество элементов таблицы и количество различных значений, минимальное и максимальное значение для каждого столбца. Статистика хранится таким образом, чтобы ее можно было комбинировать дополнительным способом, т.е. будущие вставки, а также данные из нескольких разделов могут изменять существующую статистику. Диапазон и мощность могут быть объединены. Для различных значений HMS использует представление битового массива на основе HyperLogLog++[38], которое может быть объединено без потери точности аппроксимации.

4.2 Повторная оптимизация запросов

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

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

Вторая стратегия (повторная оптимизация) основана на статистике, полученной во время выполнения. Во время планирования запроса оптимизатор оценивает размер промежуточных результатов в плане на основе статистики, полученной из HMS. Если эти оценки не точны, оптимизатор может допустить ошибки планирования, например, неправильный выбор алгоритма соединения или выделения памяти. Это, в свою очередь, может привести к снижению производительности и ошибкам выполнения. Hive собирает статистику времени выполнения для каждого оператора в плане. Если во время выполнения запроса обнаруживается какая-то вышеупомянутая проблема, запрос повторно оптимизируется с использованием статистики времени выполнения и выполняется снова.

4.3 Кэш результатов запроса

Согласованность транзакций хранилища позволяет Hive повторно использовать результаты ранее выполненного запроса, используя внутреннее состояние транзакций участвующих таблиц. Кэш запросов обеспечивает преимущества масштабируемости при работе с инструментами бизнес-аналитики, которые генерируют повторяющиеся идентичные запросы.

Каждый экземпляр HS2 хранит свой собственный компонент кэша запросов, который, в свою очередь, хранит карту от представления AST запроса до записи, содержащей местоположение результатов и информацию о моментальном снимке данных, по которым и был получен ответ на запрос. Этот компонент также отвечает за удаление устаревших записей и очистку ресурсов, используемых этими записями.

Во время компиляции запроса HS2 проверяет свой кэш, используя входной запрос AST на предварительном этапе. Неполные ссылки на таблицы в запросе разрешаются до того, как AST используется для проверки кэша, поскольку в зависимости от текущей базы данных во время выполнения запроса два запроса с одинаковым текстом могут обращаться к таблицам из разных баз данных. Если есть попадание в кэш, а таблицы, используемые запросом, не содержат новых или измененных данных, то план запроса будет состоять из одной задачи, которая будет извлекать результаты из кэша. Если запись не существует, то запрос выполняется как обычно, и результаты, сгенерированные для запроса, сохраняются в кэше, когда запрос удовлетворяет некоторым условиям, например, запрос не может содержать недетерминированные функции (rand), постоянные функции времени выполнения (current_date, current_timestamp) и т.д.

4.4. Осуществленные представления и переписывание

Традиционно одним из наиболее мощных методов, используемых для ускорения обработки запросов в хранилищах данных, является предварительное вычисление соответствующих осуществленных (реализованных) представлений [28, 31, 35-37].

Hive Apache поддерживает осуществленные представления и автоматическую перезапись запросов на основе этих реализаций. В частности, просто семантически обогащенные таблицы. Поэтому они могут храниться изначально в Hive или в других поддерживаемых системах (см. Раздел 6), могут легко использовать такие функции, как ускорение LLAP (описано в разделе 5.1). Оптимизатор использует Calcite для автоматического создания полных и частично содержащие перезаписи выражений запросов SelectProject-Join-Aggregate (SPJA) (см. Рисунок 4). Алгоритм перезаписи использует информацию об ограничениях целостности, объявленную в Hive, например, первичный ключ, внешний ключ, уникальный ключ, а не null для создания дополнительных допустимых преобразований. Алгоритм инкапсулирован в правило и запускается оптимизатором, который отвечает за принятие решения о том, следует ли использовать перезапись для ответа на запрос или нет. Обратите внимание, что если к разным частям запроса применимо несколько перезаписей, оптимизатор может в конечном итоге выбрать более одной замены представления.

Рисунок 4. Материализованное определение представления (а) образец полностью (б) и частично

Обслуживание материализованного представления. Когда данные в таблицах источников, используемых материализованным представлением, изменяются, например, вставляются новые данные или изменяются существующие данные, потребуется обновить содержимое такого представления, чтобы поддерживать его в актуальном состоянии с учетом этих изменений. В настоящее время операция перестройки для материализованного представления должна запускаться пользователем вручную с помощью инструкции REBUILD.

По умолчанию Hive пытается перестроить материализованный вид постепенно [32, 34], возвращаясь к полному перестроению, если это невозможно. Текущая реализация поддерживает инкрементное перестроение только при выполнении операций INSERT над исходными таблицами, в то время как операции UPDATE и DELETE приводят к полному перестроению материализованного представления.

Один интересный аспект заключается в том, что инкрементное выполнение зависит от самого алгоритма перезаписи. Поскольку запрос связан со снимком данных, определение материализованного представления дополняется условиями фильтрации для значения столбца WriteId каждой сканируемой таблицы (напомним раздел 3.2). Эти условия фильтров отражают моментальный снимок данных, когда материализованное представление было создано или окончательно обновлено. Когда запускается операция обслуживания, алгоритм перезаписи может произвести частично замкнутую перезапись, которая считывает материализованное представление и новые данные из исходных таблиц. Этот переписанный план, в свою очередь, преобразуется в операцию INSERT, если это материализованное представление SPJ или операцию MERGE, если это материализованное представление SPJA.

Жизненный цикл материализованного представления. По умолчанию, как только содержимое материализованного представления устаревает, оно не будет использоваться для перезаписи запроса. Однако в некоторых случаях может быть полезным согласиться на перезапись устаревших данных при обновлении материализованных представлений микро-пакетами. В этих случаях Hive позволяет пользователям периодически выполнять операцию перестройки (например, каждые 5 минут) и определять диапазон устаревания данных, разрешенный в определении материализованного представления с использованием свойство таблицы 1 (например, 10 минут).

4.5. Оптимизация совместной работы

Hive способен идентифицировать перекрывающиеся подвыражения в плане выполнения данного запроса, вычисляя их только один раз и повторно используя их результаты. Вместо запуска преобразований для поиска эквивалентных подвыражений в плане, оптимизатор совместной работы объединяет только равные части плана, аналогично другим подходам, основанным на повторном использовании и представленным в предыдущих работах [25,41]. Он применяет алгоритм повторного использования непосредственно перед выполнением: объединяются операции сканирования по одним и тем же таблицам, затем объединяются операторы плана до тех пор, пока не будет найдено различие. Решение о стратегии передачи данных для нового общего подвыражения, выходящего из объединенного выражения, остается за базовым механизмом, т.е. Apache Tez. Преимущество подхода, основанного на таком механизме, заключается в том, что он может ускорить выполнение запросов, вычисляющих одно и то же подвыражение более одного раза, без значительных затрат на оптимизацию. Тем не менее, поскольку оптимизатор совместной работы не исследует все пространство поиска эквивалентных планов, Hive может не обнаружить существующие возможности повторного использования.

4.6. Динамическое сокращение полусоединений

Сокращение полусвязи (полжизни активности соединений) – традиционный метод, используемый для уменьшения размера промежуточных результатов во время выполнения запроса [17, 20]. Оптимизация особенно полезна для БД звездной архитектуры с одной или несколькими таблицами измерений. Запросы к этим БД обычно соединяются с таблицами фактов и измерений, которые логически фильтруются с помощью предикатов в одном или нескольких столбцах. Однако эти столбцы не используются в условии соединения, следовательно, фильтр по таблице фактов не может быть создан статически. Следующий SQL-запрос показывает пример такого соединения между таблицами фактов store_sales и store_returns и таблицей измерения номенклатуры:

 SELECT ss_customer_sk, SUM(ss_sales_price) AS sum_sales
 FROM store_sales, store_returns, item
 WHERE ss_item_sk = sr_item_sk AND
 ss_ticket_number = sr_ticket_number AND
 ss_item_sk = i_item_sk AND
 i_category = 'Sports'
 GROUP BY ss_customer_sk
 ORDER BY sum_sales DESC;

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

Полусвязные редукторы вводятся оптимизатором и в операторы сканирования в плане. В зависимости от данных, Hive реализует два варианта оптимизации.

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

Индекс полусоединения. Если таблица, уменьшенная на полусоединение, не разделена столбцом соединения (индексирования), Hive может использовать значения, сгенерированные подвыражением фильтрации, чтобы, во-первых, создать условие фильтра arange с минимальными и максимальными значениями и, во-вторых, создать фильтр Блума (Bloom) со значениями, созданными подвыражением. Этот фильтр позволяет использовать любой заранее определенный объем памяти при добавлении новых элементов.

Hive заполняет полупересекающийся преобразователь данных этих двух фильтров, которые могут быть использованы для проверки на пустоту групп строк во время выполнения, например, если данные хранятся в файлах ORC [39].

5. ВЫПОЛНЕНИЕ ЗАПРОСА

Улучшения во внутренних компонентах выполнения запросов, таких как переход от Map Reduce к Apache Tez[50] и реализация операторов формата хранения на основе вектор-столбцов [39], уменьшили задержку запросов в Hive на порядки. Однако для дальнейшего улучшения времени выполнения Hive требовались дополнительные улучшения, чтобы преодолеть ограничения, присущие его первоначальной архитектуре, которая была адаптирована к длительным запросам.

Во-первых, выполнение требовало выделения контейнеров YARN при запуске, что быстро стало критическим узким местом для запросов с низкой задержкой, во-вторых, оптимизация компилятора в режиме Just in time(JIT) была не настолько эффективной, поскольку контейнеры просто уничтожались после выполнения запроса, и, в-третьих, Hive не мог использовать возможности обмена данными и кэширования как внутри, так и между запросами, что приводило к ненужным вычислительным издержкам ввода-вывода.

5.1. LLAP (Live Long and Process): «Жить долго и обрабатывать»

Длительный срок службы и процесс, известный как LLAP, – это дополнительные возможности, которые предоставляют постоянные многопоточные исполнители запросов и многопользовательский кэш в памяти для обеспечения более быстрой обработки SQL-транзакций в больших масштабах в Hive. LLAP не заменяет существующую среду выполнения, используемую Hive, такую как Tez, а скорее улучшает ее. В частности, выполнение планируется и контролируется координаторами запросов Hive прозрачно как на узлах LLAP, так и в обычных контейнерах.

Возможности ввода-вывода данных, кэширования и выполнения фрагментов запросов LLAP инкапсулированы в демонах (фоновых процессов в пользовательском пространстве). Демоны настроены на непрерывную работу в рабочих узлах кластера, что облегчает оптимизацию JIT, избегая при этом каких-либо накладных расходов при запуске. YARN используется для крупнозернистого распределения ресурсов и планирования. Он резервирует память и процессор для демона, который обрабатывает перезапуск и перемещение. Демоны не имеют состояния: каждый содержит несколько исполнителей для параллельного выполнения нескольких фрагментов запроса и локальную рабочую очередь. Сбой и восстановление упрощаются, поскольку любой узел все еще может использоваться для обработки любого фрагмента входных данных в случае сбоя демона LLAP.

Лифт ввода-вывода. Демон использует отдельные потоки для разгрузки ввода-вывода и сжатия данных, которые были определены как лифт ввода-вывода. Данные считываются пакетами и преобразуются во внутренний формат столбцов с кодировкой длины выполнения (RLE), готовый для векторной обработки. Пакет столбцов перемещается в фазу выполнения, как только он считывается, что позволяет обрабатывать предыдущие пакеты, пока готовятся следующие пакеты.

Преобразование из базового формата файла во внутренний формат данных LLAP выполняется с помощью плагинов, специфичных для каждого формата. В настоящее время LLAP поддерживает перевод из форматов ORC [10], Parquet [11] и текстовых файлов.

Лифт ввода-вывода может передавать проекции, настраиваемые предикаты и фильтры Блума в устройство чтения файлов, если они предусмотрены. Например, файлы ORC могут использовать преимущества этих структур, чтобы пропустить чтение целых групп столбцов и строк.

Кэширование данных. LLAP имеет кэш вне кучи в качестве основного буферного пула для хранения данных, поступающих в лифт ввода-вывода. Для каждого файла кэш адресуется лифтом ввода-вывода по двум измерениям: группам строк и столбцов. Набор строк и столбцов образуют блок "строка-столбец" (см. рисунок 5). Лифт ввода-вывода повторно собирает выбранную проекцию и оценивает предикаты для извлечения фрагментов, которые могут быть преобразованы в векторные пакеты для конвейерной обработки. В случае пропусков кэша, он заполняется недостающими фрагментами до выполнения восстановления. Результатом является постепенное заполнение кэша – по мере того, как пользователь перемещается по набору данных по денормализованным измерениям, что является общим шаблоном, для которого оптимизируется кэш.

LLAP кэширует метаданные и данные из входных файлов. Чтобы поддерживать действительность кэша при наличии обновлений файлов, LLAP использует уникальный идентификатор, присвоенный каждому файлу, хранящемуся в HDFS, вместе с информацией о длине файла. Это похоже на поля ETag, присутствующие в хранилищах больших двоичных объектов, таких как AWS S3 или хранилище больших двоичных объектов Azure.

Метаданные, включая информацию об индексе, кэшируются даже для данных, которые никогда не были в кэше. В частности, при первом сканировании метаданные заполняются массово, что используется для определения групп строк, которые необходимо загрузить, и для оценки предикатов перед определением пропусков кэша. Преимущество этого подхода заключается в том, что LLAP не будет загружать фрагменты, которые фактически не нужны для данного запроса, и, следовательно, позволит избежать уничтожения кэша.

Кэш постепенно изменяется, т.е. добавление новых данных в таблицу не приводит к полному аннулированию кэша. В частности, кэш получает файл только тогда, когда запрос обращается к этому файлу, что возвращает управление видимостью (значимостью) обратно в состояние транзакции запроса. Поскольку реализация ACID обрабатывает транзакции, регулируя видимость на уровне файла (см. раздел 3.2), кэш реализует представление MVCC данных, обслуживающих несколько одновременных запросов, возможно, в разных транзакционных состояниях.

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

Выполнение фрагмента запроса. Демоны LLAP выполняют произвольные фрагменты плана запроса, содержащие такие операции, как фильтрация, проецирование, преобразование данных, объединение, частичное агрегирование и сортировка. Они позволяют выполнять параллельное выполнение нескольких фрагментов из разных запросов и сеансов. Операции во фрагментах векторизованы и выполняются непосредственно во внутреннем формате RLE. Использование одного и того же формата для ввода-вывода, кэширования и выполнения сводит к минимуму работы, необходимой для взаимодействия между всеми ними.

Как упоминалось выше, LLAP не содержит собственной логики исполнения. Вместо этого его исполнители могут по существу копировать функциональность контейнеров YARN. Однако по соображениям стабильности и безопасности в LLAP принимаются только код Hive и статически определенные определяемые пользователем функции (UDF).

5.2. Улучшения в управлении рабочей нагрузкой

Менеджер рабочей нагрузки управляет доступом к ресурсам LLAP для каждого запроса, выполняемого Hive. Администратор может создавать планы ресурсов (т.е. автономные конфигурации совместного использования ресурсов) для повышения предсказуемости выполнения и совместного использования кластеров с помощью параллельных запросов, выполняемых в LLAP. Эти факторы имеют решающее значение в многопользовательских средах. Хотя в системе может быть определены несколько планов ресурсов, в данный момент времени только один из них может быть активен для данного развертывания. Планы ресурсов сохраняются Hive в HMS.

План ресурсов состоит из:

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

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

В качестве примера рассмотрим следующее определение плана ресурсов для производственного кластера:

CREATE RESOURCE PLAN daytime;
CREATE POOL daytime.bi
WITH alloc_fraction=0.8, query_parallelism=5;
CREATE POOL daytime.etl
WITH alloc_fraction=0.2, query_parallelism=20;
CREATE RULE downgrade IN daytime
WHEN total_runtime > 3000 THEN MOVE etl;
ADD RULE downgrade TO bi;
CREATE APPLICATION MAPPING visualization_app IN daytime TO bi;
 ALTER PLAN daytime SET DEFAULT POOL = etl;
 ALTER RESOURCE PLAN daytime ENABLE ACTIVATE;

Строка 1 создает план ресурсов в дневное время. Строки 2-3 создают bi пула с 80% ресурсов LLAP в кластере. Эти ресурсы могут использоваться для одновременного выполнения до 5 запросов. Аналогично, строки 4-5 создают etl пула с остальными ресурсами, которые могут использоваться для одновременного выполнения до 20 запросов. Строки 6-8 создают правило, которое перемещает запрос из bi в пул ресурсов etl, когда запрос выполняется более 3 секунд. Обратите внимание, что предыдущая операция может быть выполнена, поскольку фрагменты запроса легче упреждать по сравнению с контейнерами. Строка 9 создает сопоставление для приложения с именем interactive_bi, т.е. все запросы, выполняемые interactive_bi, изначально будут брать ресурсы из пула bi. В свою очередь, строка 10 устанавливает пул по умолчанию в etl для остальных запросов в системе. Наконец, строка 11 включает и активирует план ресурсов в кластере.

6. ОБЪЕДИНЕННАЯ СИСТЕМА ХРАНЕНИЯ (FEDERATION)

За последнее десятилетие наблюдается растущее распространение специализированных систем управления данными [54], которые стали популярными, потому что они обеспечивают более экономичную производительность для своего конкретного случая использования, чем традиционные СУБД.

В дополнение к своим собственным возможностям обработки, Hive может выступать в качестве посредника [23,27,57], поскольку он предназначен для поддержки запросов по нескольким независимым СУБД. Преимущества объединения доступа к этим системам через Hive многочисленны. Разработчики приложений могут выбрать сочетание нескольких систем для достижения желаемой производительности и функциональности, однако кодировать им необходимо только для одного интерфейса. Таким образом, приложения становятся независимыми от базовых систем данных, что обеспечивает большую гибкость при последующем изменении систем. Hive может использоваться для перемещения и преобразования данных между различными системами, устраняя необходимость в сторонних инструментах, а также в качестве посредника для глобального обеспечения контроля доступа и захвата аудиторских записей с помощью Ranger[12], Sentry [13] или помогать для обеспечения требований соответствия с помощью Atlas[2].

6.1. Обработчики хранения

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

входного формата, который описывает, как считывать данные из внешнего механизма, в том числе как разделить работу для увеличения параллелизма;
выходного формата, который описывает, как записывать данные во внешний механизм;
SERDE (сериализатор и десериализатор), который описывает, как преобразовать данные из внутреннего представления Hive во внешнее представление механизма;
крючок Metastore, механизм, подключаемый без перекомпиляции к внутренней работе Hive и определяющий методы уведомления, вызываемые как часть транзакций против HMS, например, при создании новой таблицы, поддерживаемой внешней системой или когда в такую таблицу вставляются новые строки.

Минимальная реализация используемого обработчика хранилища для чтения данных из внешней системы содержит, по крайней мере, формат ввода и десериализатор.

Как только интерфейс обработчика хранилища реализован, запрос внешней системы из Hive становится простым, и вся сложность скрыта от пользователя реализацией обработчика хранилища. Например, Apache Druid [4, 58] – это хранилище данных с открытым исходным кодом, предназначенное для запросов бизнес-аналитики (OLAP) к данным о событиях, которое широко используется для бизнес-аналитических приложений, ориентированных на пользователей. Hive предоставляет обработчик хранилища Druid, чтобы он мог использовать его эффективность для выполнения интерактивных запросов. Чтобы начать запрашивать из Hive, Druid необходимо единственное действие – зарегистрировать или создать источники данных Druid из Hive. Во-первых, если источник данных уже существует в Druid, мы можем сопоставить ему внешнюю Hive-таблицу с помощью простого оператора:

CREATE EXTERNAL TABLE druid_table_1
STORED BY 'org.apache.hadoop.hive.druid.DruidStorageHandler'
TBLPROPERTIES ('druid.datasource' = 'my_druid_source');

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

CREATE EXTERNAL TABLE druid_table_2 (
__time TIMESTAMP, dim1 VARCHAR(20), m1 FLOAT)
STORED BY 'org.apache.hadoop.hive.druid.DruidStorageHandler';

Как только источники Друидов будут доступны в Hive в виде внешних таблиц, мы сможем выполнять над ними любые разрешенные операции с таблицами.

6.2. Вычисление ускорения с использованием Calcite

Одной из наиболее мощных функций Hive является возможность использовать адаптеры Calcite [19] для переноса сложных вычислений в поддерживаемые системы, генерации запросов на языках, поддерживаемых ими.

Продолжая пример с Druid, наиболее распространенным способом Druid-запроса является использование API REST по протоколу HTTP с использованием запросов, выраженных в JSON (http://druid.io/docs/latest/querying/querying.html). Как только пользователь объявил таблицу, хранящуюся в Druid, Hive может прозрачно генерировать запросы Druid JSON из входных запросов SQL. В частности, оптимизатор применяет правила, которые соответствуют последовательности операторов в плане, и генерирует новую эквивалентную последовательность с большим количеством операций, выполняемых в Druid. Как только мы завершим оптимизацию, подмножество операторов, которые должны быть выполнены Druid, преобразуется Calcite в допустимый запрос JSON, который прикрепляется к оператору сканирования, который и будет считываться с Druid. Обратите внимание, что для обработчика хранилища при поддержке автоматически генерируемых запросов Calcite, его формат ввода должен включать логику отправки запроса во внешнюю систему (возможно разделение запроса на несколько подзапросов, которые могут выполняться параллельно) и считывание результатов запроса.

Рисунок 6. Пример федерации запросов в Hive

На сегодняшний день Hive может передавать операции на Druid и несколько «движков» поддерживаемых JDBC с использованием Calcite (Calcite может создавать SQL-запросы из выражений операторов с помощью большого количество различных диалектов). На рисунке 6 показан запрос, выполняемый над таблицей, хранящейся в Druid, и соответствующий план и запрос JSON, созданный с помощью Calcite.

7. ОЦЕНКА ЭФФЕКТИВНОСТИ РАБОТЫ

Чтобы изучить результативность нашей работы, мы оценивали Hive с течением времени, используя различные стандартные критерии. В этом разделе мы представляем краткое изложение результатов, в которых подчеркивается влияние улучшений, представленных в этом документе.

Экспериментальная установка. Эксперименты, представленные ниже, проводились на кластере из 10 узлов, соединенных сетью Ethernet 10 гигабит. Каждый узел оснащен 8-ядерным процессором Intel Xeon E5-2630 v3 с частотой 2,40 ГГц и 256 ГБ оперативной памяти, а также двумя дисками емкостью 6 ТБ для хранения HDFS и YARN.

7.1. Сравнение с предыдущими версиями

Мы провели эксперименты с использованием запросов TPC-DS в наборе данных объемом 10 ТБ. Данные хранились в таблицах ACID в HDFS с использованием формата файла ORC, а таблицы фактов были разделены по дням. Вся информация, необходимая для воспроизведения этих результатов, является общедоступной (http://github.com/hortonworks/hive-testbench/tree/hdp3). Мы сравнили две разные версии Hive [7] друг с другом:

Hive v1.2, выпущенный в сентябре 2015 года, работающий поверх Tez v0.5;
Hive v3.1, выпущенная в ноябре 2018 года, с использованием Tez v0.9 с включенным LLAP.

На рисунке 7 показано время отклика (воспринимаемое пользователем) для обеих версий; обратите внимание, используется логарифмическая ось. Для каждого запроса мы сообщаем среднее значение за три запуска с «теплым» кэшем или кэшем с некоторыми значениями, дающими попадание в кэш, ускорение. Во-первых, обратите внимание, что в Hive v1.2 может быть выполнено только 50 запросов; значения времени отклика для запросов, которые не удалось выполнить, на рисунке опущены. Причина в том, что Hive v1.2 не поддерживал операции набора, такие как ИСКЛЮЧЕНИЕ или ПЕРЕСЕЧЕНИЕ, коррелированные скалярные подзапросы с неравными условиями соединения, обозначения интервалов и порядок по невыбранным столбцам, среди других функций SQL. Для этих 50 запросов Hive v3.1 значительно превосходит предыдущую версию. В частности, Hive v3.1 быстрее в среднем в 4,6 раза и максимум в 45,5 раза (см. q58). Запросы, которые улучшились более чем в 15 раз, подчеркнуты на рисунке. Что более важно, Hive v3.1 может выполнять полный набор из 99 запросов TPC-DS благодаря улучшениям в поддержке SQL. Разница в производительности между обеими версиями настолько значительна, что совокупное время отклика для всех запросов, выполняемых Hive v3.1, по-прежнему на 15% меньше, чем время для 50 запросов в Hive v1.2. Новые функции оптимизации, такие как оптимизатор совместной работы, сами по себе имеют большое значение; например, q88 в 2,7 раза быстрее, когда он включен.

7.2. Ускорение LLAP

Чтобы наглядно продемонстрировать преимущества LLAP по сравнению с выполнением запросов при использовании контейнеров Tez, мы запускали все 99 запросов TPC-DS в Hive v3.1, используя ту же конфигурацию, но с включенным/отключенным LLAP. В таблице 1 показано совокупное время для всех запросов в эксперименте; мы можем наблюдать, что LLAP сам по себе значительно сокращает время отклика рабочей нагрузки в 2,7 раза.

Рисунок 7. Сравнение времени ответа на запрос между различными версиями Hive

Рисунок 8. Сравнение времени ответа на запрос между собственным Hive и федерацией для Druid.

7.3. «Запрос Федерации к Druid

Ниже мы проиллюстрируем значительные преимущества в производительности, которые могут быть получены в результате сочетания использования возможностей федерации Hive с материализованным представлением. Для этого эксперимента мы используем эталонный тест звездной схемы (SSB) [47] в наборе данных масштаба 1 ТБ. Тест SSB основан на TPC-Hand и предназначен для имитации процесса итеративного и интерактивного запроса хранилища данных для воспроизведения сценариев «что, если», детализации и лучшего понимания тенденций. Он состоит из одной таблицы фактов и 4 таблиц измерений, а рабочая нагрузка содержит 13 запросов, которые объединяют, агрегируют и размещают довольно жесткие фильтры измерений над различными наборами таблиц. Для эксперимента мы создаем материализованное представление, которое денормализует схему базы данных 6. Материализация хранится в Hive. Затем мы запускаем запросы в тесте, которые автоматически переписываются оптимизатором для ответа из материализованного представления. Впоследствии мы сохраняем материализованный вид в Druid v 0.12 и повторяем те же шаги. На рисунке 8 показано время отклика для каждого варианта. Обратите внимание, что Hive/Druidis в 1,6 раза быстрее, чем выполнение над материализованным представлением, хранящимся изначально в Hive. Причина в том, что Hive передает большую часть вычислений запросов Druid с использованием Calcite, и поэтому он может извлечь выгоду из того, что Druid обеспечивает меньшую задержку для этих запросов.

8. ОБСУЖДЕНИЕ

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

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

Например, корпоративные пользователи запросили внедрение гарантий ACID, чтобы разгрузить рабочие нагрузки от хранилищ данных, внедренных в их организациях задолго до Hadoop. Кроме того, недавно вступившие в силу новые правила, такие как Европейский GDPR, который дает физическим лицам право запрашивать удаления персональных данных, только подчеркнули важность этой новой функции. Неудивительно, что поддержка ACID быстро стала ключевым отличием или выбором Hive по сравнению с другими движками SQL поверх Hadoop.

Однако внедрение возможностей ACID в Hive было непростым делом, и первоначальная реализация должна была претерпеть серьезные изменения, поскольку она вводила штраф за задержку чтения по сравнению с таблицами без ACID, что было неприемлемо для пользователей. Это произошло из-за детали в оригинальном дизайне, которую мы не предвидели и которая окажет такое большое влияние на производительность в производстве – использование одного файла дельта-типа для хранения вставленных, обновленных и удаленных записей. Если рабочая нагрузка состояла из большого количества операций записи или сжатие выполнялось недостаточно часто, считывателям файлов приходилось выполнять сортировку-объединение большого количества базовых и дельта-файлов для консолидации данных, что потенциально создавало нехватку памяти. Кроме того, к этим дельта-файлам не может быть применен фильтр для пропуска чтения целых групп строк. Вторая реализация ACID, описанная в этой статье, была развернута с помощью Hive v3.1. Это решает проблемы, описанные выше, и производительность находится на уровне таблиц не ACID-типа.

Другим распространенным и частым запросом пользователей Hive с течением времени было улучшение задержки во время выполнения. Разработка LLAP началась 4 года назад для решения проблем, присущих архитектуре системы, таких как отсутствие кэширования данных или длительно работающих исполнителей. Обратная связь от первоначальных развертываний была быстро включена в систему. Например, предсказуемость была огромной проблемой в производстве из-за конкуренции за общие ресурсы кластера между пользовательскими запросами, что привело к внедрению менеджера рабочей нагрузки. В настоящее время чрезвычайно приятно видеть, как компании развертывают LLAP для обслуживания запросов по TBS данных в многопользовательских кластерах, достигая средней задержки порядка секунд.

Еще одним важным решением стала интеграция с Calcite для оптимизации запросов. Представление запросов на релевантном уровне абстракции имело решающее значение для реализации передовых алгоритмов оптимизации в Hive, что, в свою очередь, принесло системе огромные преимущества в производительности. Кроме того, интеграция была использована для простого создания запросов для объединения с другими системами. Поскольку в настоящее время организации часто используют множество систем управления данными, эта новая функция была воспринята пользователями Hive с большим энтузиазмом.

9. ДОРОЖНАЯ КАРТА

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

Статистика, полученная во время выполнения для повторной оптимизации запросов, уже сохраняется в HMS, что позволит нам передавать эту информацию в оптимизатор для получения более точных оценок, аналогичных другим системам [53, 60]. Работа над материализованными представлениями все еще продолжается, и одной из наиболее востребованных функций является внедрение советника или наставника [1,61] для Hive. Также ведутся работы по повышению производительности и стабильности LLAP, а также внедрению новых связей с другими специализированными системами, например, Kafka [9, 43].

Автономный Metastore. HMS стала важной частью Hadoop, поскольку она используется другими механизмами обработки данных, такими как Spark или Presto, для предоставления центрального хранилища информации обо всех источниках данных в каждой системе. Следовательно, растет интерес и продолжается работа по выделению Metastore из Hive и его развитию в качестве отдельного проекта. Каждая из этих систем будет иметь свой собственный каталог в Metastore и будет проще получить доступ к данным из разных систем.

Контейнерный Hive в облаке. Облачные решения для хранения данных, такие как Azure SQL DW [18], Redshift [33, 49], BigQuery [21, 46] и Snowflake [29, 51], набирают популярность в последние несколько лет. Кроме того, растет интерес к таким системам, как Kubernetes [44], которые обеспечивают сопровождение контейнеров для приложений, развернутых нечетко локально или в облаке. Модульная архитектура Hive позволяет легко изолировать его компоненты (HS2, HMS, LLAP) и запускать их в контейнерах. Текущие усилия сосредоточены на завершении работы по развертыванию Hive в коммерческих облаках с использованием Kubernetes.

10. ВЫВОДЫ

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

В этой статье мы показали, как сообщество расширило полезность системы от инструмента ETL до полноценного хранилища данных корпоративного уровня. Мы описали добавление транзакционной системы, которая хорошо подходит для изменений данных, требуемых в базах данных схемы «звезда». Мы показали основные улучшения среды выполнения, необходимые для обеспечения задержки запросов и параллелизма в области интерактивной работы. Мы также описали методы оптимизации на основе затрат, необходимые для обработки современных иерархий представлений и операций с большими данными. Наконец, мы показали, как Hive можно использовать сегодня в качестве реляционного интерфейса для нескольких систем хранения и обработки данных. Все это реализуется без какого-либо ущерба для первоначальных характеристик системы, которые сделали ее популярной. Архитектура и принципы проектирования Apache Hive доказали свою эффективность в современном аналитическом ландшафте. Мы считаем, что он будет продолжать процветать в новых средах развертывания и хранения по мере их появления, как это показывает сегодня с контейнеризацией и облаком.

Ссылки

[1] Sanjay Agrawal, Surajit Chaudhuri, and Vivek R. Narasayya. 2000. Automated Selection of Materialized Views and Indexes in SQL Databases. In PVLDB.

[2] Apache Atlas 2018. Apache Atlas: Data Governance and Metadata framework for Hadoop. http://atlas.apache.org/.

[3] Apache Calcite 2018. Apache Calcite: Dynamic data management framework. http://calcite.apache.org/.

[4] Apache Druid 2018. Apache Druid: Interactive analytics at scale. http://druid.io/. [5] Apache Flink 2018. Apache Flink: Stateful Computations over Data Streams. http://flink.apache.org/.

[6] Apache HBase 2018. Apache HBase. http://hbase.apache.org/.

[7] Apache Hive 2018. Apache Hive. http://hive.apache.org/.

[8] Apache Impala 2018. Apache Impala. http://impala.apache.org/.

[9] Apache Kafka 2018. Apache Kafka: A distributed streaming platform. http://kafka.apache.org/.

[10] Apache ORC 2018. Apache ORC: High-Performance Columnar Storage for Hadoop. http://orc.apache.org/.

[11] Apache Parquet 2018. Apache Parquet. http://parquet.apache.org/.

[12] Apache Ranger 2018. Apache Ranger: Framework to enable, monitor and manage comprehensive data security across the Hadoop platform. http://ranger.apache.org/.

[13] Apache Sentry 2018. Apache Sentry: System for enforcing fine grained role based authorization to data and metadata stored on a Hadoop cluster. http://sentry.apache.org/.

[14] Apache Spark 2018. Apache Spark: Unified Analytics Engine for Big Data. http://spark.apache.org/.

[15] Apache Tez 2018. Apache Tez. http://tez.apache.org/.

[16] Apache Thrift 2018. Apache Thrift. http://thrift.apache.org/.

[17] Peter M. G. Apers, Alan R. Hevner, and S. Bing Yao. 1983. Optimization Algorithms for Distributed Queries. IEEE Trans. Software Eng. 9, 1 (1983), 57–68. [18] Azure SQL DW 2018. Azure SQL Data Warehouse. http://azure.microsoft.com/en-us/services/sql-data-warehouse/.

[19] Edmon Begoli, Jesús Camacho-Rodríguez, Julian Hyde, Michael J. Mior, and Daniel Lemire. 2018. Apache Calcite: A Foundational Framework for Optimized Query Processing Over Heterogeneous Data Sources. In SIGMOD.

[20] Philip A. Bernstein, Nathan Goodman, Eugene Wong, Christopher L. Reeve, and James B. Rothnie Jr. 1981. Query Processing in a System for Distributed Databases (SDD-1). ACM Trans. Database Syst. 6, 4 (1981), 602–625.

[21] BigQuery 2018. BigQuery: Analytics Data Warehouse. http://cloud.google.com/bigquery/.

[22] Vinayak R. Borkar, Michael J. Carey, Raman Grover, Nicola Onose, and Rares Vernica. 2011. Hyracks: A flexible and extensible foundation for data-intensive computing. In ICDE.

[23] Francesca Bugiotti, Damian Bursztyn, Alin Deutsch, Ioana Ileana, and Ioana Manolescu. 2015. Invisible Glue: Scalable Self-Tunning MultiStores. In CIDR.

[24] Michael J. Cahill, Uwe Röhm, and Alan David Fekete. 2008. Serializable isolation for snapshot databases. In SIGMOD.

[25] Jesús Camacho-Rodríguez, Dario Colazzo, Melanie Herschel, Ioana Manolescu, and Soudip Roy Chowdhury. 2016. Reuse-based Optimization for Pig Latin. In CIKM.

[26] Paris Carbone, Asterios Katsifodimos, Stephan Ewen, Volker Markl, Seif Haridi, and Kostas Tzoumas. 2015. Apache Flink™: Stream and Batch Processing in a Single Engine. IEEE Data Eng. Bull. 38, 4 (2015), 28–38.

[27] Michael J. Carey, Laura M. Haas, Peter M. Schwarz, Manish Arya, William F. Cody, Ronald Fagin, Myron Flickner, Allen Luniewski, Wayne Niblack, Dragutin Petkovic, Joachim Thomas, John H. Williams, and Edward L. Wimmers. 1995. Towards Heterogeneous Multimedia Information Systems: The Garlic Approach. In RIDE-DOM Workshop.

[28] Surajit Chaudhuri, Ravi Krishnamurthy, Spyros Potamianos, and Kyuseok Shim. 1995. Optimizing Queries with Materialized Views. In ICDE.

[29] Benoît Dageville, Thierry Cruanes, Marcin Zukowski, Vadim Antonov, Artin Avanes, Jon Bock, Jonathan Claybaugh, Daniel Engovatov, Martin Hentschel, Jiansheng Huang, Allison W. Lee, Ashish Motivala, Abdul Q. Munir, Steven Pelley, Peter Povinec, Greg Rahn, Spyridon Triantafyllis, and Philipp Unterbrunner. 2016. The Snowflake Elastic Data Warehouse. In SIGMOD.

[30] DataNucleus 2018. DataNucleus: JDO/JPA/REST Persistence of Java Objects. http://www.datanucleus.org/.

[31] Jonathan Goldstein and Per-Åke Larson. 2001. Optimizing Queries Using Materialized Views: A practical, scalable solution. In SIGMOD.

[32] Timothy Griffin and Leonid Libkin. 1995. Incremental Maintenance of Views with Duplicates. In SIGMOD.

[33] Anurag Gupta, Deepak Agarwal, Derek Tan, Jakub Kulesza, Rahul Pathak, Stefano Stefani, and Vidhya Srinivasan. 2015. Amazon Redshift and the Case for Simpler Data Warehouses. In SIGMOD.

[34] Ashish Gupta and Inderpal Singh Mumick. 1995. Maintenance of Materialized Views: Problems, Techniques, and Applications. IEEE Data Eng. Bull. 18, 2 (1995), 3–18.

[35] Himanshu Gupta. 1997. Selection of Views to Materialize in a Data Warehouse. In ICDT.

[36] Himanshu Gupta, Venky Harinarayan, Anand Rajaraman, and Jeffrey D. Ullman. 1997. Index Selection for OLAP. In ICDE.

[37] Venky Harinarayan, Anand Rajaraman, and Jeffrey D. Ullman. 1996. Implementing Data Cubes Efficiently. In SIGMOD.

[38] Stefan Heule, Marc Nunkesser, and Alexander Hall. 2013. HyperLogLog in practice: algorithmic engineering of a state of the art cardinality estimation algorithm. In EDBT.

[39] Yin Huai, Ashutosh Chauhan, Alan Gates, Günther Hagleitner, Eric N. Hanson, Owen O’Malley, Jitendra Pandey, Yuan Yuan, Rubao Lee, and Xiaodong Zhang. 2014. Major technical advancements in apache hive. In SIGMOD.

[40] Michael Isard, Mihai Budiu, Yuan Yu, Andrew Birrell, and Dennis Fetterly. 2007. Dryad: distributed data-parallel programs from sequential building blocks. In EuroSys.

[41] Alekh Jindal, Shi Qiao, Hiren Patel, Zhicheng Yin, Jieming Di, Malay Bag, Marc Friedman, Yifung Lin, Konstantinos Karanasos, and Sriram Rao. 2018. Computation Reuse in Analytics Job Service at Microsoft. In SIGMOD.

[42] Marcel Kornacker, Alexander Behm, Victor Bittorf, Taras Bobrovytsky, Casey Ching, Alan Choi, Justin Erickson, Martin Grund, Daniel Hecht, Matthew Jacobs, Ishaan Joshi, Lenni Kuff, Dileep Kumar, Alex Leblang, Nong Li, Ippokratis Pandis, Henry Robinson, David Rorke, Silvius Rus, John Russell, Dimitris Tsirogiannis, Skye Wanderman-Milne, and Michael Yoder. 2015. Impala: A Modern, Open-Source SQL Engine for Hadoop. In CIDR.

[43] Jay Kreps, Neha Narkhede, and Jun Rao. 2011. Kafka : a Distributed Messaging System for Log Processing. In NetDB.

[44] Kubernetes 2018. Kubernetes: Production-Grade Container Orchestration. http://kubernetes.io/.

[45] Per-Åke Larson, Cipri Clinciu, Campbell Fraser, Eric N. Hanson, Mostafa Mokhtar, Michal Nowakiewicz, Vassilis Papadimos, Susan L. Price, Srikumar Rangarajan, Remus Rusanu, and Mayukh Saubhasik. 2013. Enhancements to SQL server column stores. In SIGMOD.

[46] Sergey Melnik, Andrey Gubarev, Jing Jing Long, Geoffrey Romer, Shiva Shivakumar, Matt Tolton, and Theo Vassilakis. 2010. Dremel: Interactive Analysis of Web-Scale Datasets. In PVLDB. [47] Patrick E. O’Neil, Elizabeth J. O’Neil, Xuedong Chen, and Stephen Revilak. 2009. The Star Schema Benchmark and Augmented Fact Table Indexing. In TPCTC.

[48] Presto 2018. Presto: Distributed SQL query engine for big data. http://prestodb.io/.

[49] Redshift 2018. Amazon Redshift: Amazon Web Services. http://aws.amazon.com/redshift/.

[50] Bikas Saha, Hitesh Shah, Siddharth Seth, Gopal Vijayaraghavan, Arun C. Murthy, and Carlo Curino. 2015. Apache Tez: A Unifying Framework for Modeling and Building Data Processing Applications. In SIGMOD.

[51] Snowflake 2018. Snowflake: The Enterprise Data Warehouse Built in the Cloud. http://www.snowflake.com/.

[52] Mohamed A. Soliman, Lyublena Antova, Venkatesh Raghavan, Amr El-Helw, Zhongxian Gu, Entong Shen, George C. Caragea, Carlos Garcia-Alvarado, Foyzur Rahman, Michalis Petropoulos, Florian Waas, Sivaramakrishnan Narayanan, Konstantinos Krikellas, and Rhonda Baldwin. 2014. Orca: a modular query optimizer architecture for big data. In SIGMOD.

[53] Michael Stillger, Guy M. Lohman, Volker Markl, and Mokhtar Kandil. 2001. LEO - DB2’s LEarning Optimizer. In PVLDB.

[54] Michael Stonebraker and Ugur Çetintemel. 2005. "One Size Fits All": An Idea Whose Time Has Come and Gone. In ICDE.

[55] Ashish Thusoo, Joydeep Sen Sarma, Namit Jain, Zheng Shao, Prasad Chakka, Ning Zhang, Suresh Anthony, Hao Liu, and Raghotham Murthy. 2010. Hive - a petabyte scale data warehouse using Hadoop. In ICDE.

[56] Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas, Sharad Agarwal, Mahadev Konar, Robert Evans, Thomas Graves, Jason Lowe, Hitesh Shah, Siddharth Seth, Bikas Saha, Carlo Curino, Owen O’Malley, Sanjay Radia, Benjamin Reed, and Eric Baldeschwieler. 2013. Apache Hadoop YARN: yet another resource negotiator. In SOCC.

[57] Gio Wiederhold. 1992. Mediators in the Architecture of Future Information Systems. IEEE Computer 25, 3 (1992), 38–49.

[58] Fangjin Yang, Eric Tschetter, Xavier Léauté, Nelson Ray, Gian Merlino, and Deep Ganguli. 2014. Druid: a real-time analytical data store. In SIGMOD.

[59] Matei Zaharia, Mosharaf Chowdhury, Michael J. Franklin, Scott Shenker, and Ion Stoica. 2010. Spark: Cluster Computing with Working Sets. In USENIX HotCloud.

[60] Mohamed Zaït, Sunil Chakkappen, Suratna Budalakoti, Satyanarayana R. Valluri, Ramarajan Krishnamachari, and Alan Wood. 2017. Adaptive Statistics in Oracle 12c. In PVLDB.

[61] Daniel C. Zilio, Jun Rao, Sam Lightstone, Guy M. Lohman, Adam J. Storm, Christian Garcia-Arellano, and Scott Fadden. 2004. DB2 Design Advisor: Integrated Automatic Physical Database Design. In PVLDB.

Apache Hive: от модели распределённых вычислений MapReduce компании Google до Big Data – хранилища больших данных