ACID-транзакции в Hive

Константин Алпашкин

Hive поддерживает транзакции ACID, которые обладают следующими свойствами:

  • Атомарность (atomicity). Гарантирует, что транзакция либо выполняется полностью, либо не выполняются вообще. Другими словами, транзакции неделимы, и если одна часть транзакции не выполняется, вся транзакция откатывается, оставляя систему в исходном состоянии.

  • Консистентность (consistency). Обеспечивает целостность базы данных как до, так и после выполнения транзакции. Транзакции обеспечивают целостность данных и наложенных на них констрейнтов.

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

  • Надежность (durability). В момент коммита транзакции ее изменения становятся постоянными и не теряются даже в случае системного сбоя.

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

Дельта-файлы и сжатие

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

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

Существует два типа сжатий:

  • Минорное (minor). Объединяет несколько дельта-файлов в один.

  • Мажорное (major). Переписывает один или несколько дельта-файлов и базовый файл в новый базовый файл. Этот процесс гораздо более ресурсоемкий, чем минорное сжатие.

Помимо автоматических запусков процесса сжатия вы также можете запустить процесс вручную с помощью команды COMPACT.

Активация транзакций ACID

По умолчанию транзакции Hive ACID отключены. Для работы с транзакционными таблицами необходимо установить следующие параметры сервиса Hive в ADCM (Clusters → <clusterName> → Services → Hive → Primary configuration → hive-site.xml):

  • hive.support.concurrency=true;

  • hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager (установлено по умолчанию).

Использование транзакционных таблиц

Создание таблицы

Для создания транзакционной таблицы с полной поддержкой ACID необходимо следующее:

  • Транзакции Hive ACID включены в ADCM.

  • Создаваемая таблица разделена на бакеты.

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

  • При создании таблицы устанавливается свойство TBLPROPERTIES ("transactional"="true").

Пример создания транзакционной таблицы:

CREATE TABLE `transactions`
    (`txn_id` int,
    `acc_id` int,
    `txn_amount` decimal(10,2),
    `txn_date` date)
    CLUSTERED BY (acc_id) INTO 3 BUCKETS (1)
    STORED AS ORC (2)
    TBLPROPERTIES ("transactional"="true"); (3)
1 Разделение новой таблицы на бакеты.
2 Использование ORC в качестве формата хранения.
3 Установка свойства для создания транзакционной таблицы.

Используя DESCRIBE FORMATTED <table_name>, можно проверить, что новая таблица была создана как транзакционная. Ниже приведена строка вывода, подтверждающая это.

...
| transactional                                     | true                                               |
...

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

This command is not allowed on an ACID table <table_name> with a non-ACID transaction manager

Преобразование в ACID-таблицу

Вы можете преобразовать обычную нетранзакционную таблицу Hive в таблицу со свойствами ACID. Главное требование — исходная таблица Hive должна быть типа managed, а в качестве формата хранения должен использоваться ORC. Например:

ALTER TABLE <non-acid_table> SET TBLPROPERTIES ("transactional"="true");

Чтобы наделить таблицу типа external свойствами ACID, для начала ее необходимо сделать managed-таблицей. Например:

ALTER TABLE <external_table_name> SET TBLPROPERTIES("EXTERNAL"="false")
ПРИМЕЧАНИЕ
Если таблица Hive создана как ACID-таблица, ее обратная конвертация в не-ACID таблицу невозможна.

Вставка и обновление данных

Команды записи или обновления данных для транзакционной таблицы не отличаются от команд для обычных таблиц. Синтаксис команд INSERT/UPDATE/DELETE — стандартный, например:

INSERT INTO transactions VALUES
(1, 1002, 10.00, '2023-01-01'),
(2, 1001, 100.50, '2023-01-02'),
(3, 1003, 50.00, '2023-01-03');

Главной особенностью при записи в транзакционную таблицу является то, что Hive создает дельта-файлы в директории HDFS warehouse. Чтобы просмотреть дельта-файлы, созданные в результате выполнения приведенного выше запроса INSERT, выполните команду:

$ hdfs dfs -ls /apps/hive/warehouse/transactions/delta_0000001_0000001_0000
РЕКОМЕНДАЦИЯ
Числа в имени директории delta_0000001_0000001_0000 означают ID транзакции.

Вывод:

Found 3 items
-rw-r--r--   3 hive hadoop          1 2024-03-28 17:21 /apps/hive/warehouse/transactions/delta_0000001_0000001_0000/_orc_acid_version (1)
-rw-r--r--   3 hive hadoop        906 2024-03-28 17:21 /apps/hive/warehouse/transactions/delta_0000001_0000001_0000/bucket_00000 (2)
-rw-r--r--   3 hive hadoop        959 2024-03-28 17:21 /apps/hive/warehouse/transactions/delta_0000001_0000001_0000/bucket_00002 (2)
1 Файл содержит версию ORC.
2 Файлы хранят данные, разделенные по бакетам.

Специальные команды транзакций

Ниже показаны несколько команд, специфичных для работы с транзакциями Hive.

COMPACT

С помощью следующей команды вы можете вручную запустить операцию минорного/мажорного сжатия:

ALTER TABLE <table_name> COMPACT 'major|minor';

Выполнение этой команды помещает операцию сжатия в очередь и выводит идентификатор операции сжатия.

...
Compaction enqueued with id {id}

Чтобы увидеть список всех операций сжатия, выполните команду:

SHOW COMPACTIONS;

Пример вывода:

+---------------+-----------+----------------+------------+--------+------------+-----------+-------------+---------------+--------------+
| compactionid  |  dbname   |    tabname     |  partname  |  type  |   state    | workerid  |  starttime  |   duration    | hadoopjobid  |
+---------------+-----------+----------------+------------+--------+------------+-----------+-------------+---------------+--------------+
| CompactionId  | Database  | Table          | Partition  | Type   | State      | Worker    | Start Time  | Duration(ms)  | HadoopJobId  |
| 1             | default   | transactions   |  ---       | MINOR  | initiated  |  ---      |  ---        |  ---          | None         |
+---------------+-----------+----------------+------------+--------+------------+-----------+-------------+---------------+--------------+

SHOW TRANSACTIONS

Данная команда выводит список всех транзакций в системе, например:

+-----------------+--------------------+----------------+----------------------+-------+--------------------------------+
|      txnid      |       state        |  startedtime   |  lastheartbeattime   | user  |              host              |
+-----------------+--------------------+----------------+----------------------+-------+--------------------------------+
| Transaction ID  | Transaction State  | Started Time   | Last Heartbeat Time  | User  | Hostname                       |
| 10              | ABORTED            | 1711646536000  | 1711646536000        | hive  | ka-adh-2.ru-central1.internal  |
| 29              | OPEN               | 1711970603000  | 1711970603000        | hive  | ka-adh-2.ru-central1.internal  |
+-----------------+--------------------+----------------+----------------------+-------+--------------------------------+

SHOW LOCKS

С помощью этой команды вы можете увидеть все блокировки (locks) для таблицы или партиции. Пример вывода показан ниже.

+----------+------------------+----------------+------------+-------------+---------------+--------------+-----------------+-----------------+----------------+-------+--------------------------------+----------------------------------------------------+
|  lockid  |     database     |     table      | partition  | lock_state  |  blocked_by   |  lock_type   | transaction_id  | last_heartbeat  |  acquired_at   | user  |            hostname            |                     agent_info                     |
+----------+------------------+----------------+------------+-------------+---------------+--------------+-----------------+-----------------+----------------+-------+--------------------------------+----------------------------------------------------+
| Lock ID  | Database         | Table          | Partition  | State       | Blocked By    | Type         | Transaction ID  | Last Heartbeat  | Acquired At    | User  | Hostname                       | Agent Info                                         |
| 41.1     | _dummy_database  | _dummy_table   | NULL       | ACQUIRED    |               | SHARED_READ  | 50              | 0               | 1711992560000  | hive  | ka-adh-2.ru-central1.internal  | hive_20240401172919_fbb1738c-5736-458f-956e-d89ea7d3a727 |
| 41.2     | default          | transactions2  | NULL       | ACQUIRED    |               | SHARED_READ  | 50              | 0               | 1711992560000  | hive  | ka-adh-2.ru-central1.internal  | hive_20240401172919_fbb1738c-5736-458f-956e-d89ea7d3a727 |
+----------+------------------+----------------+------------+-------------+---------------+--------------+-----------------+-----------------+----------------+-------+--------------------------------+----------------------------------------------------+

Очистка данных ACID

Поддержка ACID-транзакций предполагает запись вспомогательных данных в системные таблицы Hive Metastore. Такие данные отражают отмененные и закоммиченные транзакции и в основном сохраняются в таблицу TXNS. В производственной среде с высокой нагрузкой подобные записи могут быстро накапливаться и негативно влиять на производительность Hive Metastore.

Для предотвращения роста объема системных данных Hive предоставляет механизм очистки (AcidTxnCleanerService), который периодически удаляет соответствующие записи в системных таблицах Metastore. Этот механизм по умолчанию отключен и может быть активирован с помощью настроек ACID Housekeeping в ADCM (hive-site.xml → metastore.housekeeping.threads.on).

Очистка в режиме HA

Механизм очистки работает с несколькими компонентами Hive Metastore, установленными в кластере ADH (режим высокой доступности). В этом случае есть два способа, чтобы указать, какой именно компонент Hive Metastore должен выполнять физическое удаление данных. Поведение зависит от значения параметра metastore.housekeeping.leader.election, а именно:

  • host. Компонент Hive Metastore, который будет выполнять очистку, указывается явно с помощью FQDN. Требует установки параметра metastore.housekeeping.leader.hostname.

  • lock. Лидер Hive Metastore выбирается с помощью ZooKeeper lock. Требует установки параметра metastore.housekeeping.leader.lock.namespace (например, hive_acid).

Ограничения

Существует несколько ограничений для использования транзакций Hive:

  • В отличие от традиционных транзакций SQL, операции BEGIN, COMMIT и ROLLBACK не поддерживаются. Все транзакции Hive работают в режиме auto-commit.

  • ORC — единственный поддерживаемый формат хранения.

  • Чтобы использовать транзакции ACID, таблица Hive должна быть разбита на бакеты.

  • Внешние (external) таблицы не могут быть транзакционными.

  • Чтение/запись в ACID-таблицу из не ACID-сессии запрещены. Другими словами, для работы с ACID-таблицами должен использоваться менеджер транзакций Hive org.apache.hadoop.hive.ql.lockmgr.DbTxnManager.

  • Менеджеры блокировок In-memory и ZooKeeper не совместимы с транзакциями.

  • Использование Oracle в качестве базы данных для Metastore, а также свойства datanucleus.connectionPoolingType=BONECP может приводить к периодическим ошибкам "No such lock …​" и "No such transaction …​". В этом случае рекомендуется установить значение datanucleus.connectionPoolingType=DBCP.

  • Операции LOAD DATA не поддерживаются для транзакционных таблиц.

Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней