Hive连接器

概述

Hive连接器支持查询存储在Hive数据仓库中的数据。Hive由三个部分组成：

• 存储在HDFS或Amazon S3中的不同格式的数据文件。
• 有关数据文件如何映射到模式和表的元数据。该元数据存储在MySQL等数据库中，通过Hive元存储服务访问。
• 一种称为HiveQL的查询语言。这种查询语言在分布式计算框架（如MapReduce或Tez）上执行。

openLooKeng只使用前两个部分：数据和元数据。不使用HiveQL或Hive的任何部分执行环境。

支持的文件类型

Hive连接器支持以下文件类型：

• ORC
• Parquet
• Avro
• RCFile
• SequenceFile
• JSON
• 文本

为了在使用Hive 3.x时能够对Avro表和CSV文件实现一级支持，需要在Hive元存储配置文件hive-site.xml中添加如下属性定义：

<property>
     <!-- https://community.hortonworks.com/content/supportkb/247055/errorjavalangunsupportedoperationexception-storage.html -->
     <name>metastore.storage.schema.reader.impl</name>
     <value>org.apache.hadoop.hive.metastore.SerDeStorageSchemaReader</value>
 </property>

配置

Hive连接器支持Apache Hadoop 2.x及其衍生发行版，包括Cloudera CDH 5和Hortonworks Data Platform (HDP)。

用以下内容创建etc/catalog/hive.properties，以将hive-hadoop2连接器挂载为hive目录，将example.net:9083替换为Hive元存储Thrift服务的正确主机和端口：

connector.name=hive-hadoop2
hive.metastore.uri=thrift://example.net:9083

多Hive集群

可以根据需要创建任意多的目录，因此，如果有额外的Hive集群，只需添加另一个不同的名称的属性文件到etc/catalog中（确保它以.properties结尾）。例如，如果将属性文件命名为sales.properties，openLooKeng将使用配置的连接器创建一个名为sales的目录。

HDFS配置

对于基本设置，openLooKeng自动配置HDFS客户端，不需要任何配置文件。在某些情况下，例如使用联邦HDFS或NameNode高可用性时，需要指定额外的HDFS客户端选项，以便访问HDFS集群。要指定选项，添加hive.config.resources属性来引用HDFS配置文件：

hive.config.resources=/etc/hadoop/conf/core-site.xml,/etc/hadoop/conf/hdfs-site.xml

如果设置需要，只指定附加的配置文件。同时建议减少配置文件，使其具有所需的最小属性集，因为附加属性可能导致问题。

所有openLooKeng节点上必须存在配置文件。如果你正在引用现有的Hadoop配置文件，请确保将其复制到任何没有运行Hadoop的openLooKeng节点。

HDFS用户名和权限

在openLooKeng中对Hive表执行任何CREATE TABLE或CREATE TABLE AS语句之前，需要确保openLooKeng访问HDFS的用户能够访问Hive的仓库目录。Hive的仓库目录由hive-site.xml中的配置变量hive.metastore.warehouse.dir指定，默认值为/user/hive/warehouse。

在不使用带HDFS的Kerberos时，openLooKeng会使用openLooKeng进程的操作系统用户来访问HDFS。例如，如果openLooKeng作为nobody运行，则openLooKeng将作为nobody访问HDFS。可以通过在openLooKeng JVM配置中设置HADOOP_USER_NAME系统属性来覆盖此用户名，用适当的用户名替换hdfs_user：

-DHADOOP_USER_NAME=hdfs_user

hive用户通常可行，因为Hive通常随hive用户启动，并且该用户可以访问Hive仓库。

无论何时修改openLooKeng访问HDFS的用户时，请移除HDFS上的/tmp/openlookeng-*，/tmp/presto-*，/tmp/hetu-*，因为新用户可能无法访问现有的临时目录。

访问Kerberos身份验证保护的Hadoop集群

HDFS和Hive元存储都支持Kerberos认证。但是，目前还不支持通过票据缓存进行Kerberos身份验证。

Hive连接器安全需要的属性在Hive配置属性表中列出。有关Hive连接器中安全选项的更详细讨论，请参阅Hive安全配置部分。

Hive配置属性

Hive Thrift 元存储配置属性说明

AWS Glue目录配置属性

Amazon S3配置

Hive连接器支持读写存储在S3中的表。这可以通过使用S3前缀而不是HDFS前缀的表或数据库位置来实现。

openLooKeng对URI前缀s3://、s3n://和s3a://使用自己的S3文件系统。

S3配置属性

S3凭据

如果使用EMR或其他工具在Amazon EC2上运行openLooKeng，强烈建议将hive.s3.use-instance-credentials设置为true并使用IAM Rolesfor EC2来控制对S3的访问。如果是这种情况，则需要为EC2实例分配一个IAM角色，以授予对存储在希望使用的S3桶中的数据的适当访问权限。还可以通过hive.s3.iam-role配置一个IAM角色，访问任何S3桶时都使用该角色。这比在hive.s3.aws-access-key和hive.s3.aws-secret-key设置中设置AWS访问和密钥要明确得多，而且还允许EC2定期自动轮换凭据，而无需进行任何额外工作。

自定义S3凭据提供程序

可以通过将Hadoop配置属性presto.s3.credentials-provider设置为自定义AWS凭据提供程序实现的完全限定类名，来配置自定义S3凭据提供程序。此类必须实现AWSCredentialsProvider接口，并提供将java.net.URI和Hadoop org.apache.hadoop.conf.Configuration作为参数的双参数构造函数。自定义凭据提供程序可用于提供来自STS的临时凭据（使用STSSessionCredentialsProvider）、基于IAM角色的凭据（使用STSAssumeRoleSessionCredentialsProvider）或特定用例的凭据（例如，桶/用户特定凭据）。此Hadoop配置属性必须在hive.config.resources Hive连接器属性引用的Hadoop配置文件中进行设置。

调优属性

当与S3通信时，以下调优属性会影响openLooKeng S3文件系统所使用客户端的行为。这些参数中的大多数会影响与AmazonS3Client关联的ClientConfiguration对象上的设置。

S3数据加密

openLooKeng支持使用S3托管密钥的服务器端加密和使用Amazon KMS或软件插件管理AES加密密钥的客户端加密在S3中读取和写入加密数据。

使用S3服务器端加密（在Amazon文档中称为SSE-S3）,S3基础架构负责所有加密和解密工作（客户端SSL除外，假设已将hive.s3.ssl.enabled设置为true）.S3还管理所有的加密密钥。将hive.s3.sse.enabled设置为true来开启此功能。

使用S3客户端加密，S3存储加密数据，加密密钥在S3基础架构之外进行管理。数据加解密由openLooKeng完成，不在S3基础架构中。在这种情况下，可以通过使用AWS KMS或自己的密钥管理系统来管理加密密钥。使用AWS KMS进行密钥管理时，将hive.s3.kms-key-id设置为KMS密钥的UUID。还需要授予AWS凭据或EC2 IAM角色使用给定密钥的权限。

若要使用自定义的加密密钥管理系统，将hive.s3.encryption-materials-provider设置为实现来自AWS Java SDK的EncryptionMaterialsProvider接口的类的完全限定名。该类必须通过类路径对Hive连接器进行访问，并且必须能够与自定义密钥管理系统进行通信。如果该类也实现了来自Hadoop Java API的org.apache.hadoop.conf.Configurable接口，那么Hadoop配置将在对象实例创建之后，请求提供或检索任何加密密钥之前传入。

S3 Select下推

S3 Select下推功能，可以将下推投影（SELECT）和谓词（WHERE）处理下推到S3 Select。使用S3 Select下推，openLooKeng只从S3中而不是从所有S3对象检索所需的数据，从而降低了延迟和网络利用率。

S3 Select是否适合我的工作负载？

S3 Select下推的性能取决于查询过滤的数据量。筛选大量行应能获得更好的性能。如果查询没有过滤任何数据，则下推可能不会增加任何附价值，并且用户将因S3 Select请求而被收费。因此，建议对使用S3 Select和不使用S3 Select的工作负载进行基准测试以查看使用S3 Select是否适合工作负载。默认情况下，S3 Select下推是禁用的。应在适当的基准测试和成本分析之后在生产中启用。有关S3 Select请求成本的更多信息，请参阅Amazon S3云存储定价。

使用以下准则来确定S3 Select是否适合工作负载：

• 查询过滤掉了超过一半的原始数据集。
• 查询过滤谓词使用的列具有openLooKeng和S3 Select支持的数据类型。S3 Select下推不支持TIMESTAMP、REAL和DOUBLE数据类型。建议使用十进制数据类型来表示数值数据。有关S3选择所支持的数据类型的详细信息，见数据类型文档。
• Amazon S3和Amazon EMR集群之间的网络连接具有良好的传输速度和可用带宽。Amazon S3 Select不压缩HTTP响应，因此对于压缩的输入文件，响应大小可能会增加。

注意事项和限制

• 只支持CSV格式存储的对象。对象可以是未压缩的，也可以是用gzip或bzip2压缩的。
• 不支持“AllowQuotedRecordDelimiters”属性。如果指定了此属性，则查询失败。
• 不支持使用客户提供的加密密钥（SSE-C）进行Amazon S3服务器端加密和客户端加密。
• S3 Select下推不能替代ORC、Parquet等列存或压缩文件格式。

开启S3 Select下推

可以通过s3_select_pushdown_enabled Hive会话属性或hive.s3select-pushdown.enabled配置属性启用S3 Select下推。会话属性将重写配置属性，允许按查询启用或禁用。

了解和调优最大连接数

openLooKeng可以使用其原生的S3文件系统或EMRFS。使用原生FS时，最大连接是通过hive.s3.max-connections配置属性配置的。当使用EMRFS时，最大连接是通过fs.s3.maxConnections Hadoop配置属性配置的。

S3 Select下推在访问Amazon S3进行谓词操作时绕过文件系统。在这种情况下，hive.s3select-pushdown.max-connections的值决定了工作节点允许这些操作的最大客户端连接数。

如果工作负载遇到错误“等待池中连接超时”，增加hive.s3select-pushdown.max-connections以及正在使用的文件系统的最大连接配置的值。

Google云存储配置

Hive连接器可以通过gs:// URI前缀访问存储在GCS中的数据。请参阅hive-gcs-tutorial以获取详细说明。

GCS配置属性

ORC缓存配置

Hive连接器缓存ORC文件数据，以提供更好的性能并减少查询时延。工作节点将数据缓存在本地内存中。当启用ORC缓存时,工作节点会缓存ORC文件的尾部信息,stripe的页脚(stripe-footer)信息, 行索引(row-index), 布隆过滤(bloom-filter)信息。然而, 工作节点只会缓存特定的数据行,这些行应该与cache table语句中谓语词相匹配。

ORC缓存属性

TTL: 是指自最后一次读写cache到现在的时间间隔。如后文所讲, 在写cache阶段会周期性执行时间过期验证,在读cache的时也会触发过期验证。

表统计信息

Hive连接器在写入数据时，总是收集基本的统计信息（numFiles、numRows、rawDataSize、totalSize），默认还会收集列级统计信息：

更新表和分区统计信息

如果查询较复杂且包括联接大型数据集，那么在表/分区上运行/sql/analyze可能会通过收集有关数据的统计信息来提高查询性能。

在分析分区表时，可以通过可选partitions属性指定要分析的分区，该属性是一个数组，包含分区键的值，其顺序与在表模式中所声明的顺序一致：

ANALYZE table_name WITH (
    partitions = ARRAY[
        ARRAY['p1_value1', 'p1_value2'],
        ARRAY['p2_value1', 'p2_value2']])

此查询将收集具有键p1_value1, p1_value2和p2_value1, p2_value2的两个分区的统计信息。

Hive ACID支持

openLooKeng支持Hive事务表上的ACID事务（INSERT，UPDATE，DELETE）。

使用Hive连接器创建事务表

要支持ACID事务，必须满足以下前提条件：

1. 创建表应启用了“事务性”属性。
2. 格式为ORC。参见限制

示例：

CREATE TABLE hive_acid_table (
    id int,
    name string )
  WITH (format='ORC', transactional=true);

对事务表执行INSERT

对于事务和非事务表，插入操作在最终用户的角度来看保持不变。

示例：

INSERT INTO hive_acid_table
  VALUES
     (1, 'foo'),
     (2, 'bar');

对事务表执行UPDATE

对事务表的UPDATE操作允许用户更新与WHERE子句匹配的特定行的列。

示例：

UPDATE hive_acid_table
  SET name='john'
  WHERE id=2;

上述示例将值为2的列id的行的列name的值更新为john。

UPDATE前的SELECT结果：

lk:default> SELECT * FROM hive_acid_table;
id | name
----+------
  2 | bar
  1 | foo
(2 rows)

UPDATE后的SELECT结果

lk:default> SELECT * FROM hive_acid_table;
 id | name
----+------
  2 | john
  1 | foo
(2 rows)

对事务表执行DELETE

事务表的DELETE操作允许用户更新与WHERE子句匹配的特定行的列。

示例：

DELETE FROM hive_acid_table
  WHERE id=2;

以上示例删除了值为2的列id的行。

DELETE前的SELECT结果：

lk:default> SELECT * FROM hive_acid_table;
 id | name
----+------
  2 | john
  1 | foo
(2 rows)

DELETE后的SELECT结果：

lk:default> SELECT * FROM hive_acid_table;
 id | name
----+------
  1 | foo
(1 row)

对事务表执行VACUUM

Hive将所有事务（INSERT/UPDATE/DELETE）保存在单独的delta目录中，以进行簿记。DELETE事务并不物理删除存储数据中的旧行，而是在新文件中将其标记为已删除。UPDATE使用分片更新机制更新数据。由于这些原因，对表的读操作需要读取许多文件，这增加了额外的开销。所有这些增量文件都需要合并，以获得合并后的数据，从而加快处理速度。openLooKeng中的VACUUM操作负责合并这些增量文件。

VACUUM操作在Hive中转换为compaction。在Hive中有两种类型的压缩，Major和Minor。

VACUUM和Hive compaction的映射关系如下：

• VACUUM FULL转换为Major compaction。

• VACUUM转换为Minor compaction。

VACUUM

VACUUM TABLE hive_acid_table;

上述操作触发对hive_acid_table的VACUUM操作，将合并所有delta目录到单个delta目录。一旦操作达到RUNNING状态，它将继续异步运行，解除客户机的阻塞。

说明： 目前没有命令来获取异步VACUUM的结果，但是可以从UI监视。

VACUUM FULL

Full Vacuum合并delta目录中的所有事务，并创建只包含最终结果的base目录。 该操作将永久删除已删除的行。因此，在读取期间，总的数据读取将非常少，因此提高了性能。

VACUUM TABLE hive_acid_table
  FULL;

上述操作触发对hive_acid_table的VACUUM FULL，一旦操作到达RUNNING状态，将继续异步运行，解除客户机的阻塞。

对分区表的VACUUM操作

如果表已分区，那么VACUUM操作可以单独对特定的分区进行操作，而不是对表的所有分区一起操作。

示例：创建分区表并INSERT数据：

CREATE TABLE hive_acid_table_partitioned (
    id int,
    name string,
    class int) 
WITH (format='ORC', transactional=true, partitioned_by=ARRAY['class']);
INSERT INTO hive_acid_table_partitioned
  VALUES
    (1, 'foo', 5),
    (2, 'bar', 10);

对特定分区class=5的VACUUM操作运行如下：

VACUUM TABLE hive_acid_table_partitioned
   FULL
   PARTITION 'class=5';

如果未指定PARTITION 'class=5'，则VACUUM操作对所有分区进行，如果任何一个分区中的任何故障都会导致整个操作失败。

AND WAIT选项

默认情况下，VACUUM操作以异步方式运行，即一旦查询达到RUNNING状态，客户端将不再等待操作完成。

为VACUUM操作添加AND WAIT选项，使客户端同步等待VACUUM操作完成。

以上带AND WAIT选项的示例如下：

VACUUM TABLE hive_acid_table
  FULL
  AND WAIT;

VACUUM TABLE hive_acid_table
  AND WAIT;

VACUUM TABLE hive_acid_table_partitioned
  FULL
  PARTITION 'class=5'
  AND WAIT;

模式演进

Hive允许表中的分区与表具有不同的模式。当分区已经存在（使用原始列类型）后，更改表的列类型时会发生这种情况。Hive连接器通过允许与Hive相同的转换来支持此特性：

• 从varchar转换到tinyint、smallint、integer和bigint以及相反
• 从real转换到double
• 对整数范围进行扩大转换，如从tinyint转换到smallint

任何转换失败都会产生null，这与Hive的情况相同。例如，将字符串'foo'转换为数字，或将字符串'1234'转换为tinyint（其最大值为127）。

Avro模式演进

openLooKeng支持对Avro存储格式的Hive表进行查询和操作，Avro存储格式的模式是基于Avro模式文件/原文设置。也可以在openLooKeng中创建表，从本地或远程的HDFS/Web服务器中的有效Avro模式文件中推断模式。

要指定Avro模式应该用于解释表数据，必须使用avro_schema_url表属性。模式可以远程放置在HDFS、（例如avro_schema_url = 'hdfs://user/avro/schema/avro_data.avsc'）、S3（例如avro_schema_url = 's3n:///schema_bucket/schema/avro_data.avsc'）、web服务器（例如avro_schema_url = 'http://example.org/schema/avro_data.avsc'）以及本地文件系统中。模式所在的URL必须可以从Hive元存储和openLooKeng协调节点/工作节点访问。

在openLooKeng中使用avro_schema_url创建的表的行为与设置了avro.schema.url或avro.schema.literal的Hive表的行为相同。

示例：

CREATE TABLE hive.avro.avro_data (
   id bigint
 )
WITH (
   format = 'AVRO',
   avro_schema_url = '/usr/local/avro_data.avsc'
)

如果指定了avro_schema_url，则DDL中列出的列（上述示例中的id）将被忽略。表模式将与Avro模式文件中的模式匹配。在进行任何读取操作之前，都会访问Avro模式，因此查询结果将反映模式中的任何变化。因此，openLooKeng利用了Avro的向后兼容性能力。

如果Avro模式文件中的表的模式发生变化，仍然可以使用新模式读取旧数据。Avro模式中新增/重命名的字段必须有默认值。

Schema的演进行为如下：

• 新模式中增加的列：当表使用新模式时，使用旧模式创建的数据将产生默认值。
• 新模式中移除的列：使用旧模式创建的数据将不再输出已移除列的数据。
• 列在新的模式中被重命名：这等价于移除列并添加新列，当表使用新模式时，使用旧模式创建的数据将产生默认值。
• 更改新模式中的列类型：如果Avro或Hive连接器支持类型强制，则发生转换。不兼容的类型将引发错误。

限制

设置了avro_schema_url时，不支持以下操作：

• 不支持CREATE TABLE AS。
• 不支持在CREATE TABLE中使用分区（partitioned_by）列和分桶（bucketed_by）列。
• 不支持修改列的ALTER TABLE命令。

操作步骤

• system.create_empty_partition(schema_name, table_name, partition_columns, partition_values)

  在指定表中创建一个空分区。

• system.sync_partition_metadata(schema_name, table_name, mode)

  检查并更新元存储中的分区列表。有三种方式：

  • ADD：添加文件系统中存在而元存储中不存在的分区。
  • DROP：删除元存储中存在而文件系统中不存在的分区。
  • FULL：同时执行ADD和DROP。

示例

Hive连接器支持查询和操作Hive表和模式（数据库）。虽然有些不常用的操作需要直接通过Hive来执行，但是大多数操作都可以通过openLooKeng来执行。

创建一个名为web的新的Hive模式，用于存放名为my-bucket的S3桶中的表：

CREATE SCHEMA hive.web
WITH (location = 's3://my-bucket/')

在web模式中创建一个名为page_views的新的Hive表，以ORC文件格式存储，按日期和国家分区，按用户分为到50个桶。（注意Hive要求分区列是表的最后一列）：

从page_views表中删除分区：

DELETE FROM hive.web.page_views
WHERE ds = DATE '2016-08-09'
  AND country = 'US'

给page_views表添加空分区：

CALL system.create_empty_partition(
    schema_name => 'web',
    table_name => 'page_views',
    partition_columns => ARRAY['ds', 'country'],
    partition_values => ARRAY['2016-08-09', 'US']);

查询page_views表：

SELECT * FROM hive.web.page_views

列出page_views表的分区：

SELECT * FROM hive.web."page_views$partitions"

创建名为request_logs的Hive外部表，表名指向S3中已存在的数据：

CREATE TABLE hive.web.request_logs (
  request_time timestamp,
  url varchar,
  ip varchar,
  user_agent varchar
)
WITH (
  format = 'TEXTFILE',
  external_location = 's3://my-bucket/data/logs/'
)

对request_logs表进行统计：

ANALYZE hive.web.request_logs;

将s3://替换为gs://后，此处展示的示例应该适用于Google云存储。

清理

删除外部表request_logs。这只会删除表的元数据。引用的数据目录未删除：

DROP TABLE hive.web.request_logs

删除模式：

DROP SCHEMA hive.web

性能调优说明：

INSERT

• 可以通过配置更多数量的写入器任务配置来调整大量加载插入操作，如CREATE TABLE AS或INSERT INTO TABLE SELECT COL1, COL2 FROM QUERY。

  SET SESSION task_writer_count=<num>;
  #Note: `num' is default number of local parallel table writer jobs per worker, must be a power of 2.
  #Recommended value: 50% of the total cpu cores available in the worker node can be given here

  每个工作节点有多个写入器会确保更高的数据消耗，但会导致每个分区生成多个文件；大量小文件对于读取操作来说是不理想的。

  推荐管理员/开发人员可以使用以下方式来确保生成较少文件：-

  • 对于AArch64：

    • 使用vacuum操作unify合并每个分区中由多个文件写入创建的多个文件，这样在读取期间调度拆分会更快。

      VACUUM TABLE catalog_sales FULL UNIFY;

      Vacuum unify之前：Insert通过每个写入器为给定分区生成多个文件：

      

      Vacuum unify之后：VACUUM FULL UNIFY命令，一个给定分区的所有文件小文件统一为一个文件。

      

  • 对于Intel x86：

    • 以下会话参数可以通过添加一个计划节点来在工作节点间重排记录，确保每个分区只创建一个文件；这样，只有指定的工作节点才写入特定的分区。

      SET SESSION hive.write_partition_distribution=true
      #Default: false

  • Hive元存储超时

    大分区表包含过多分区，导致任务超时。大量分区可能需要更多时间来加载和与元存储缓存同步，因此，为了在更大规模存储中获得更好的性能，建议相应地调整’hive.metastore-timeout’参数。

    应在Hive配置文件中设置以下属性。

    hive.metastore-timeout=<TimeWithUnit>;
    #说明：'TimeWithUnit'为时间，单位为秒或分钟。 
    #默认值：10s（其中's'表示秒）
    #推荐值：对于大分区表中的操作，值可为60s或更大，需要根据数据量进行配置。此处显示的值仅供参考，建议根据实际情况进行调整。

• 并行元存储操作

  应对用户会话设置以下参数。

  SET SESSION hive.metastore-client-service-threads = 4
  #Default: 4
  #Recommended: The number of running HMS service.

  根据许多并行HMS操作可以调用的线程池的数量，这将减少获取分区的总时间。

  说明：另外，集群中可以添加多个Hive元存储服务，这些服务将以轮询的方式访问，从而保证更佳的Hive元存储负载。

• 直接删除整个分区

  如果删除请求是针对分区列，那么该列将被直接删除，因此将使用元数据调用删除分区。删除增量文件将不会在该分区上创建，因为该分区的整个数据将被删除。

  应在应用会话级别上设置以下参数：

  delete_transactional_table_direct=true
  #Default: false

  使用:

  DELETE FROM table_test WHERE partition_column >= (SELECT max(done_paritions) FROM tracking_table);

  说明：

  a）直接删除只支持>,>=,<,<=运算符。

  b）删除整个分区时，delete命令的输出不能打印删除的记录数。

已知问题

• 在运行并发查询（包括选择、更新、删除、清空）或清空清理时，部分查询可能会由于冲突而失败。读查询也可能失败并报错“FileNotFoundException”。这些场景都是Hive ACID utils的bug导致，但不会造成数据丢失。此外，重新运行读取/选择查询也会成功。

Hive连接器限制

• 只有当表是非事务性，WHERE子句匹配整个分区时，才支持DELETE。对于事务型表，WHERE子句可以是任何条件。

• Hive元数据库不支持schema重命名，导致ALTER SCHEMA使用失败。

• openLooKeng只支持Hive 3.x版本的Hive表的ACID事务。

• openLooKeng目前仅支持ORC格式的事务表。