Hive 中的 ACID 和 Transaction，事务表支持

2021-02-28 本文已影响0人高斯纯牛奶

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 中，以解决以下用例：

流式传输数据。许多用户使用诸如Apache Flume，Apache Storm或Apache Kafka之类的工具来将数据流式传输到 Hadoop 集群中。虽然这些工具可以每秒数百行或更多行的速度写入数据，但 Hive 只能每隔 15 分钟到一个小时添加一次分区。频繁添加分区会很快导致表中大量的分区。这些工具可以将数据流式传输到现有分区中，但是这会使 Reader 感到脏读(也就是说，他们将在开始查询后看到写入的数据)，并将许多小文件留在目录中，这将给 NameNode 带来压力。使用此新功能，将支持该用例，同时允许 Reader 获得一致的数据视图并避免过多的文件。
尺寸变化缓慢。在典型的星型模式数据仓库中，维度表随时间缓慢变化。例如，零售商将开设新 Store，需要将其添加到 Store 表中，或者现有 Store 可能会更改其平方英尺或某些其他跟踪的 Feature。这些更改导致插入单个记录或更新记录(取决于所选策略)。从 0.14 开始，Hive 可以支持此功能。
数据重述。有时发现收集的数据不正确，需要更正。或者，数据的第一个实例可能与以后提供的完整数据近似(占报告服务器的 90％)。或业务规则可能会要求某些 Transaction 由于后续 Transaction 而重述(例如，在进行购买后，Client 可以购买会员资格，因此有权享受折扣价，包括之前的购买价格)。或者，Contract 可能会要求用户在终止关系后删除其 Client 的数据。从 Hive 0.14 开始，可以通过* INSERT，UPDATE 和 DELETE *支持这些用例。
使用SQL MERGE语句进行批量更新。

Limitations

尚不支持* BEGIN ， COMMIT 和 ROLLBACK *。所有语言操作都是自动提交的。该计划是在将来的版本中支持它们。
此第一个版本仅支持ORC 文件格式。构建该功能的目的是使事务可以被任何存储格式使用，该存储格式可以确定如何将更新或删除应用于基本记录(基本上具有显式或隐式的行 ID)，但是到目前为止，集成工作仅针对 ORC。
默认情况下，事务配置为关闭。请参阅下面的Configuration部分，以讨论需要设置哪些值来进行配置。
表格必须为bucketed才能使用这些功能。同一系统中未使用事务和 ACID 的表无需存储。外部表不能成为 ACID 表，因为外部表的更改超出了压缩器(HIVE-13175)的控制范围。
不允许从非 ACID 会话读取/写入 ACID 表。换句话说，必须将 Hive 事务 Management 器设置为 org.apache.hadoop.hive.ql.lockmgr.DbTxnManager 才能使用 ACID 表。
目前，仅支持快照级别隔离。当给定查询开始时，将为它提供一致的数据快照。不支持脏读，已提交读，可重复读或可序列化。引入 BEGIN 的目的是在事务期间支持快照隔离，而不仅仅是单个查询。可以根据用户请求添加其他隔离级别。
现有的 ZooKeeper 和内存锁 Management 器与事务不兼容。无意解决此问题。有关如何存储事务锁的讨论，请参见下面的Basic Design。
ACID 表不支持使用 ALTER TABLE 进行模式更改。 HIVE-11421正在跟踪它。在 1.3.0/2.0.0 中修复。
将 Oracle 用作 Metastore DB 和“ datanucleus.connectionPoolingType = BONECP”可能会生成间歇性的“无此锁..”和“无此 Transaction...”错误。在这种情况下，建议设置“ datanucleus.connectionPoolingType = DBCP”。
事务表不支持LOAD DATA...语句。 (直到HIVE-16732，这才被正确实施)

Streaming APIs

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

Hive HCatalog 流 API
Hive 流 API(自配置单元 3 起)
HCatalog 流式突变 API(在 Hive 2.0.0 和更高版本中可用)

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

Grammar Changes

INSERT ... VALUES，UPDATE 和 DELETE *已添加到 SQL 语法中，从Hive 0.14开始。有关详情，请参见LanguageManual DML。

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

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

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

SHOW LOCKS *命令已更改，以提供有关与事务关联的新锁的信息。如果您使用的是 ZooKeeper 或内存中的锁 Management 器，您将注意到此命令的输出没有差异。有关详情，请参见Show Locks。

新的选项已添加到* 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

hive.support.concurrency –是
hive.enforce.bucketing – true(自Hive 2.0起不是必需的)
hive.exec.dynamic.partition.mode –非严格
hive.txn.manager – org.apache.hadoop.hive.ql.lockmgr.DbTxnManager

服务器端(Metastore)

hive.compactor.initiator.on –正确(有关更多详细信息，请参见下表)
hive.compactor.worker.threads –在至少一个 Thrift Metastore 服务实例上为正数

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

事务的新配置参数

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

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