Hive 中的 ACID 和 Transaction

Hive 3 Warning

由 Hive 3 之前的 Hive 版本创建的任何事务表都需要在每个分区上运行 Major Compaction，然后才能升级到 3.0. 更准确地说，自上次重大压缩以来已对其执行任何更新/删除/合并语句的任何分区，都必须进行另一次重大压缩。在将 Hive 升级到 Hive 3 之后，此分区上再也不会发生更新/删除/合并。

什么是 ACID，为什么要使用它？

ACID 代表数据库事务的四个 Feature：原子性(操作完全成功或失败，它不会留下部分数据)，一致性(一旦应用程序执行了操作，则该操作的结果在每个后续操作中对它可见)； Isolation(一个用户的不完整操作不会对其他用户造成意外的副作用)和耐用性(一旦操作完成，即使在机器或系统出现故障的情况下，它也会保留下来)。长期以来，人们一直期望数据库系统将这些 Feature 作为其事务功能的一部分。

直到 Hive 0.13，在分区级别提供了原子性，一致性和持久性。可以通过打开一种可用的锁定机制(ZooKeeper或在内存中)来提供隔离。通过在Hive 0.13中添加事务，现在可以在行级别提供完整的 ACID 语义，因此一个应用程序可以添加行，而另一个应用程序可以从同一分区读取数据而不会互相干扰。

具有 ACID 语义的事务已添加到 Hive 中，以解决以下用例：

Limitations

Streaming APIs

Hive 提供用于流式数据提取和流式突变的 API：

“流变异”文档的Background部分提供了这两个 API 的比较。

Grammar Changes

Hive 的 DDL 中已添加了一些新命令以支持 ACID 和事务，此外还修改了一些现有的 DDL。

已添加新命令* SHOW TRANSACTIONS *，有关详细信息，请参见Show Transactions。

已添加新命令* SHOW COMPACTIONS *，有关详细信息，请参见Show Compactions。

新的选项已添加到* ALTER TABLE ，以请求压缩表或分区。通常，用户不需要请求压缩，因为系统会检测到对它们的需求并启动压缩。但是，如果表的压缩已关闭或用户想在系统不选择的时间压缩表，则 ALTER TABLE 可用于启动压缩。有关详情，请参见修改表/分区契约。这将排队请求压缩并返回。要查看压缩的进度，用户可以使用 SHOW COMPACTIONS *。

已添加新命令* ABORT TRANSACTIONS *，有关详细信息，请参见Abort Transactions。

Basic Design

HDFS 不支持对文件的就地更改。面对编写者附加到用户正在读取的文件时，它也不提供读取一致性。为了在 HDFS 之上提供这些功能，我们遵循了其他数据仓库工具中使用的标准方法。表或分区的数据存储在一组基本文件中。新记录，更新和删除存储在增量文件中。将为更改表或分区的每个事务(或就 Flume 或 Storm 这样的流代理而言，每个事务)创建一组新的增量文件。在读取时，阅读器合并基本文件和增量文件，并在读取时应用所有更新和删除。

基本目录和增量目录

以前，分区(如果未分区表，则为表)的所有文件都位于单个目录中。进行这些更改后，使用 ACID 感知编写器编写的任何分区(或表)都将具有基本文件目录和每组增量文件目录。这是未分区表“ t”的外观：

表“ t”的文件系统布局

hive> dfs -ls -R /user/hive/warehouse/t;
drwxr-xr-x   - ekoifman staff          0 2016-06-09 17:03 /user/hive/warehouse/t/base_0000022
-rw-r--r--   1 ekoifman staff        602 2016-06-09 17:03 /user/hive/warehouse/t/base_0000022/bucket_00000
drwxr-xr-x   - ekoifman staff          0 2016-06-09 17:06 /user/hive/warehouse/t/delta_0000023_0000023_0000
-rw-r--r--   1 ekoifman staff        611 2016-06-09 17:06 /user/hive/warehouse/t/delta_0000023_0000023_0000/bucket_00000
drwxr-xr-x   - ekoifman staff          0 2016-06-09 17:07 /user/hive/warehouse/t/delta_0000024_0000024_0000
-rw-r--r--   1 ekoifman staff        610 2016-06-09 17:07 /user/hive/warehouse/t/delta_0000024_0000024_0000/bucket_00000

Compactor

Compactor 是在 Metastore 中运行的一组后台进程，以支持 ACID 系统。它由发起者，工作者，清理者，AcidHouseKeeperService 和其他一些人组成。

Delta 文件压缩

随着操作修改表，将创建越来越多的增量文件，并且需要对其进行压缩以保持足够的性能。压缩有两种类型，次要压缩和主要压缩。

所有压缩都是在后台完成的，并且不会阻止并发读取和写入数据。压缩后，系统将 await，直到所有旧文件的读取器都已完成，然后再删除旧文件。

Initiator

该模块负责发现哪些表或分区需要进行压缩。应该在 Metastore 中使用hive.compactor.initiator.on启用它。下表“事务的新配置参数”表中有* .threshold 格式的多个属性，这些属性控制何时创建压缩任务以及执行哪种压缩类型。每个压缩任务处理 1 个分区(如果表未分区，则处理整个表)。如果给定分区的连续压缩失败次数超过 hive.compactor.initiator.failed.compacts.threshold，则自动压缩调度将为此分区停止。有关更多信息，请参见配置参数表。

Worker

每个 Worker 处理一个压缩任务。压缩是具有以下格式的名称的 MapReduce 作业：\ -compactor-<db>。\ 。\ 。每个工作人员将作业提交到群集(如果定义，则通过hive.compactor.job.queue)，然后 await 作业完成。 hive.compactor.worker.threads确定每个 Metastore 中的 Worker 数量。蜂房仓库中的 Worker 总数决定了并发压实的最大数量。

Cleaner

此过程是在压缩后确定不再需要增量文件后将其删除的过程。

AcidHouseKeeperService

此过程将查找在hive.txn.timeout时间内未引起您欢迎的 Transaction 并将其中止。系统假定启动事务的 Client 端停止心跳崩溃，并且应该释放其锁定的资源。

SHOW COMPACTIONS

此命令显示有关当前正在运行的压缩和最近的压缩历史(可配置的保留期限)的信息。自HIVE-12353以来，此历史记录显示可用。

有关此命令输出的更多信息，另请参见LanguageManual DDL#ShowCompactions，有关影响该命令输出的配置属性，请参见NewConfigurationParametersforTransactions/Compaction 历史记录。系统将保留每种类型的最后 N 个条目：失败，成功，尝试(其中每种类型的 N 可配置)。

Transaction/Lock Manager

添加了一个称为“事务 Management 器”的新逻辑实体，其中合并了以前的“数据库/表/分区锁 Management 器”概念(hive.lock.manager，默认值为 org.apache.hadoop.hive.ql.lockmgr.zookeeper.ZooKeeperHiveLockManager) 。现在，事务 Management 器还负责事务锁的 Management。默认的 DummyTxnManager 模拟旧的 Hive 版本的行为：没有事务，并使用 hive.lock.manager 属性为表，分区和数据库创建锁 Management 器。新添加的 DbTxnManager 使用 DbLockManagerManagementHive 元存储中的所有锁/事务(在服务器发生故障时，事务和锁是持久的)。这意味着启用事务后，不再存在先前的 ZooKeeper 锁定行为。为避免 Client 端死于事务并留下 Transaction 或锁悬而未决，系统会定期将心跳 signal 从锁持有者和 Transaction 发起者发送到元存储。如果在配置的时间内未收到心跳，则锁或事务将中止。

从Hive 1.3.0开始，DbLockManger 将 continue 尝试获取锁的时间长度可以通过[hive.lock.numretires](http://Configuration Properties#hive.lock.numretires)和[hive.lock.sleep.between.retries](http://Configuration Properties#hive.lock.sleep.between.retries)进行控制。当 DbLockManager 无法获取锁(由于存在竞争锁)时，它将退出并在一定时间后重试。为了支持运行时间短的查询，而不是同时淹没 metastore，DbLockManager 将在每次重试后将 await 时间加倍。初始退避时间为 100 毫秒，并受 hive.lock.sleep.between.retries 限制。 hive.lock.numretries 是它将重试给定锁定请求的总次数。因此，获取锁的调用将阻塞的总时间(给定值 100 重试和 60s 睡眠时间)为(100ms 200ms 400ms ... 51200ms 60s 60s ... 60s)= 91m：42s：300ms。

此锁定 Management 器使用的锁定更多details。

请注意，DbTxnManager 使用的锁 Management 器将获取所有表的锁，即使没有“ transactional = true”属性的表也是如此。默认情况下，对非事务表的插入操作将获得排他锁，从而阻止其他插入和读取。尽管从技术上讲是正确的，但这与 Hive 传统的工作方式(即不带锁 Management 器)背道而驰。为了向后兼容，提供了[hive.txn.strict.locking.mode](http://Configuration Properties#hive.txn.strict.locking.mode)(请参见下表)，这将使此锁 Management 器在非事务表上的插入操作上获取共享锁。这将恢复以前的语义，同时仍提供锁 Management 器的优点，例如防止在读取表时将其丢弃。请注意，对于事务表，insert 始终获取共享锁，因为这些表在存储层实现了 MVCC 体系结构，即使在存在并发修改操作的情况下，也能够提供强大的读取一致性(快照隔离)。

Configuration

至少，必须正确设置以下配置参数以在 Hive 中打开事务支持：

Client Side

服务器端(Metastore)

以下各节列出了影响 Hive 事务和压缩的所有配置参数。另请参见上方的Limitations和下方的Table Properties。

事务的新配置参数

许多新的配置参数已添加到系统以支持事务。

Configuration key Values Location Notes
hive.txn.manager 默认值： org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager

Transaction 所需的值： org.apache.hadoop.hive.ql.lockmgr.DbTxnManager		 Client /
HiveServer2		 DummyTxnManager 复制 Hive-0.13 之前的行为，不提供任何事务。
hive.txn.strict.locking.mode 默认值： true Client 端/ HiveServer2 在严格模式下，非 ACID 资源使用标准的 R/W 锁定语义，例如 INSERT 将获得排他锁。在非严格模式下，对于非 ACID 资源，INSERT 将仅获取共享锁，这允许两次并发写入同一分区，但仍允许锁 Management 器在写入表时(从Hive 2.2.0开始)防止 DROP TABLE 等。 。
hive.txn.timeout Default: 300 Client/
HiveServer2/
如果 Client 端未发送心跳，则 Metastore		 声明的事务终止时间(以秒为单位)。对于所有组件/服务，此属性具有相同的值至关重要。5
hive.txn.heartbeat.threadpool.size 默认值： 5 Client/
HiveServer2		 用于心跳的线程数(从Hive1.3.0 和 2.0.0开始)。
hive.timedout.txn.reaper.start 默认值： 100s Metastore 元存储开始后(从Hive 1.3.0开始)运行第一个收割者的时间延迟(中止超时事务的过程)。控制上面的 AcidHouseKeeperServcie。
hive.timedout.txn.reaper.interval 默认值： 180s Metastore 时间间隔，描述收割者(终止超时事务的过程)运行的频率(自Hive 1.3.0)。控制上面的 AcidHouseKeeperServcie。
hive.txn.max.open.batch 默认值： 1000 Client 一次调用 open_txns()。1 可获取的最大 Transaction 数。
hive.max.open.txns 默认值： 100000 HiveServer2 /
Metastore		 最大未结 Transaction 数。如果当前未完成的 Transaction 达到此限制，则将来的未完成 Transaction 请求将被拒绝，直到数量低于该限制。 (截至Hive1.3.0 和 2.1.0。)
hive.count.open.txns.interval 默认值： 1s HiveServer2 /
 Metastore 检查两次未结 Transaction 的时间(以Hive1.3.0 和 2.1.0为准)。
hive.txn.retryable.sqlex.regex 默认值：“”(空字符串) HiveServer2 /
Metastore		 Comm 分隔的 SQL 状态，错误代码和可重试 SQLException 错误消息的正则表达式模式列表，适用于 Hive Metastore 数据库(自Hive1.3.0 和 2.1.0开始)。
有关示例，请参见Configuration Properties。
hive.compactor.initiator.on Default: false
Transaction 所需的值： true(仅适用于 Thrift Metastore 服务的一个实例)		 Metastore 是否在此 Metastore 实例上运行启动程序线程和清理程序线程。在Hive 1.3.0之前，至关重要的是，仅在一个独立的 metastore 服务实例(尚未实施)上启用此功能。
从Hive 1.3.0开始，可以在任意数量的独立 metastore 实例上启用此属性。
hive.compactor.worker.threads Default: 0
Transaction 所需的值：> 0 在 Thrift Metastore 服务的至少一个实例上		 Metastore 在此 Metastore 实例上运行多少个压缩程序工作线程。2
hive.compactor.worker.timeout 默认值： 86400 Metastore 以秒为单位的时间，在此时间之后，将宣布压缩作业失败并重新进行压缩排队。
hive.compactor.cleaner.run.interval 默认：5000 Metastore 两次清洁线程运行之间的时间(以毫秒为单位)。 (Hive 0.14.0及更高版本。)
hive.compactor.check.interval 默认值： 300 Metastore 两次检查之间的时间(以秒为单位)，以查看是否需要压缩任何表或分区。3
hive.compactor.delta.num.threshold 默认值： 10 Metastore 表或分区中将触发较小压缩的增量目录数。
hive.compactor.delta.pct.threshold 默认值： 0.1 Metastore 相对于基数的增量文件的百分比(分数)大小，这将触发主要压缩。 1 = 100％，因此默认值为 0.1 = 10％。
hive.compactor.abortedtxn.threshold 默认值： 1000 Metastore 涉及给定表或分区的，将触发主要压缩的异常中止事务数。
hive.compactor.max.num.delta 默认：500 Metastore 压缩程序将在单个作业中尝试处理的最大增量文件数(自Hive 1.3.0以来)。4
hive.compactor.job.queue 默认：“”(空字符串) 元 Store 用于指定将向其提交压缩作业的 Hadoop 队列的名称。设置为空字符串以让 Hadoop 选择队列(从Hive 1.3.0开始)。
压缩历史记录
配置单元的历史记录保留成功 *默认值：3 * Metastore 要保留在历史 Logging 的成功压缩条目数(每个分区)。
hive.compactor.history.retention。失败 *默认值：3 * Metastore 要保留在历史 Logging 的失败压缩条目数(每个分区)。
配置单元.compactor.history.retention.attempted *默认值：2 * Metastore 在历史 Logging 保留的尝试压缩条目数(每个分区)。
hive.compactor.initiator.failed.compacts.threshold *默认值：2 * Metastore 给定分区连续失败的压缩次数，此后启动器将停止尝试自动调度压缩。仍然可以使用ALTER TABLE来启动压缩。一旦手动启动的压缩成功，自动启动的压缩将恢复。请注意，该值必须小于 hive.compactor.history.retention.failed。
hive.compactor.history.reaper.interval *默认值：2m * Metastore 控制清除压实历史记录的过程运行的频率。

1 hive.txn.max.open.batch 控制流代理(例如 Flume 或 Storm)同时打开多少个事务。然后，流代理将该数量的条目写入单个文件(每个 Flume 代理或 Storm bolt)。因此，增加此值将减少流代理创建的增量文件的数量。但是，这也增加了 Hive 在任何给定时间必须跟踪的未清事务数量，这可能会对读取性能产生负面影响。

2 工作线程产生 MapReduce 作业进行压缩。他们自己不做压实。一旦确定需要压缩，增加工作线程的数量将减少压缩表或分区所花费的时间。随着更多 MapReduce 作业将在后台运行，这还将增加 Hadoop 群集上的后台负载。每次压缩可以一次处理一个分区(如果未分区，则可以处理整个表)。

3 减小此值将减少需要压缩的表或分区开始压缩的时间。但是，检查是否需要压缩需要对自上次大型压缩以来已对其执行事务的每个表或分区多次调用 NameNode。因此，减小此值将增加 NameNode 上的负载。

4 如果压缩程序检测到大量增量文件，它将首先运行几次部分较小的压缩(当前按 Sequences 进行)，然后执行实际要求的压缩。

5 如果该值不相同，则可以确定活动 Transaction“超时”，因此中止。这将导致出现诸如“没有此类 Transaction...”，“没有此类锁...”之类的错误。

要为 INSERT，UPDATE，DELETE 设置的配置值

除了上面列出的新参数之外，还需要设置一些现有参数以支持* INSERT ... VALUES，UPDATE 和* DELETE *。

Configuration key 必须设置为
hive.support.concurrency true(默认为 false)
hive.enforce.bucketing true(默认为 false)(自Hive 2.0起不是必需的)
hive.exec.dynamic.partition.mode 非严格(默认为严格)

为压缩设置的配置值

如果您的系统中的数据不是由 Hive 用户(即 Hive Metastore 的运行用户)拥有的，则 Hive 将需要以拥有数据的用户身份运行才能执行压缩。如果您已经设置了 HiveServer2 来模拟用户，那么唯一要做的额外工作就是确保 Hive 有权从运行 Hive 元存储库的主机上模拟用户。这是通过将主机名添加到 Hadoop 的core-site.xml文件中的hadoop.proxyuser.hive.hosts来完成的。如果尚未执行此操作，则需要将 Hive 配置为充当代理用户。这要求您为运行 Hive Metastore 的用户设置密钥表，并将hadoop.proxyuser.hive.hosts和hadoop.proxyuser.hive.groups添加到 Hadoop 的core-site.xml文件中。请参阅适用于您的 Hadoop 版本的安全模式下的 Hadoop 文档(例如，对于 Hadoop 2.5.1，它位于安全模式下的 Hadoop)。

Table Properties

如果要在 ACID 写入(插入，更新，删除)中使用表，则必须在该表上设置表属性“ transactional=true”，从Hive 0.14.0开始。注意，一旦通过 TBLPROPERTIES(“ transactional” =“ true”)将一个表定义为 ACID 表，就无法将其转换回非 ACID 表，即，更改 TBLPROPERTIES(“ transactional” =“ false”)为不允许。另外，在运行任何查询之前，必须在 hive-site.xml 或会话开始时将hive.txn.manager设置为 org.apache.hadoop.hive.ql.lockmgr.DbTxnManager。没有这些，插入将以旧样式完成；在 HIVE-11716 之前，禁止进行任何更新和删除。由于 HIVE-11716 不允许在没有 DbTxnManager 的 ACID 表上进行操作。但是，这不适用于 Hive 0.13.0.

如果表所有者不希望系统自动确定何时进行压缩，则可以设置表属性“ NO_AUTO_COMPACTION”。这将阻止所有自动压缩。手动压缩仍然可以使用修改表/分区契约语句完成。

如 Hive 数据定义语言的Create Table和更改表属性部分中所述，在创建或更改表时，使用 TBLPROPERTIES 子句设置表属性。在 Hive 发行版 0.x 和 1.0 中，“ transactional”和“ NO_AUTO_COMPACTION”表属性区分大小写，但从版本 1.1.0(HIVE-8308)开始，它们不区分大小写。

从Hive1.3.0 和 2.1.0开始，可以通过 TBLPROPERTIES 设置更多与压缩有关的选项。既可以通过CREATE TABLE在表级设置，也可以通过更改表/拼音版在请求级设置。这些用于覆盖仓库/表范围的设置。例如，要覆盖 MR 属性以影响压缩作业，可以在 CREATE TABLE 语句中或通过 ALTER TABLE 显式启动压缩时添加“ compactor。\ =<value>”。将在压缩 MR 作业的 JobConf 上设置“<mr property name> =<value>”。同样，“ tblprops。\ =<value>”可用于设置/覆盖由集群上运行的代码解释的任何表属性。最后，“ compactorthreshold。\ =<value>”可用于覆盖上面的“事务的新配置参数”表中以“ .threshold”结尾的属性，并控制系统何时触发压缩。例子：

示例：在表级别的 TBLPROPERTIES 中设置压缩选项

CREATE TABLE table_name (
  id                int,
  name              string
)
CLUSTERED BY (id) INTO 2 BUCKETS STORED AS ORC
TBLPROPERTIES ("transactional"="true",
  "compactor.mapreduce.map.memory.mb"="2048",     -- specify compaction map job properties
  "compactorthreshold.hive.compactor.delta.num.threshold"="4",  -- trigger minor compaction if there aremore than 4 delta directories
  "compactorthreshold.hive.compactor.delta.pct.threshold"="0.5" -- trigger major compaction if the ratio ofsize of delta files to
                                                                   -- size of base filesis greater than 50%
);

示例：在请求级别的 TBLPROPERTIES 中设置压缩选项

ALTER TABLE table_name COMPACT 'minor'
   WITH OVERWRITE TBLPROPERTIES ("compactor.mapreduce.map.memory.mb"="3072");  -- specify compaction map job properties
ALTER TABLE table_name COMPACT 'major'
   WITH OVERWRITE TBLPROPERTIES ("tblprops.orc.compress.size"="8192");         -- change any other Hive table properties

讲座和演讲

Eugene Koifman 在2017 年 Dataworks 峰会，美国加利福尼亚圣何塞的 Hive 中的事务性操作

美国加利福尼亚 State 圣何塞 DataWorks Summit 2018-涵盖 Hive 3 和 ACID V2 功能

首页