Hive

自 2.1.3 版本开始，PALO 支持对 Hive 的 DDL 和 DML 操作。用户可以直接通过 PALO 在 Hive 中创建库表，并将数据写入到 Hive 表中。通过该功能，用户可以通过 PALO 对 Hive 进行完整的数据查询和写入操作，进一步帮助用户简化湖仓一体架构。

本文介绍在 PALO 中支持的 Hive 操作，语法和使用须知。

提示： 使用前，请先设置： set global enable_nereids_planner = true; set global enable_fallback_to_original_planner = false; 从老版本升级上来的集群，这些参数可能有变化。

元数据创建与删除

Catalog

• 创建

  Plain Text 复制

  1CREATE CATALOG [IF NOT EXISTS] hive PROPERTIES (
  2    "type"="hms",
  3    "hive.metastore.uris" = "thrift://172.21.16.47:7004",
  4    "hadoop.username" = "hadoop",
  5    "fs.defaultFS" = "hdfs://172.21.16.47:4007"
  6);

  注意，如果需要通过 PALO 创建 Hive 表或写入数据，需要在 Catalog 属性中显式增加 fs.defaultFS 属性。如果创建 Catalog 仅用于查询，则该参数可以省略。

• 删除

  Plain Text 复制

  1DROP CATALOG [IF EXISTS] hive;

  删除 Catalog 并不会删除 hive 中的任何库表信息。仅仅是在 PALO 中移除了对这个 Hive 集群的映射。

Database

• 创建

  可以通过 SWITCH 语句切换到对应的 Catalog 下，执行 CREATE DATABASE 语句：

  Plain Text 复制

  1SWITCH hive;
  2CREATE DATABASE [IF NOT EXISTS] hive_db;

  也可以使用全限定名创建，或指定 location，如：

  Plain Text 复制

  1CREATE DATABASE [IF NOT EXISTS] hive.hive_db;
  2    
  3CREATE DATABASE [IF NOT EXISTS] hive.hive_db
  4PROPERTIES ('location'='hdfs://172.21.16.47:4007/path/to/db/');

  之后可以通过 SHOW CREATE DATABASE 命令可以查看 Database 的 Location 信息：

  Plain Text 复制

  1mysql> SHOW CREATE DATABASE hive_db;
  2+----------+---------------------------------------------------------------------------------------------+
  3| Database | Create Database                                                                             |
  4+----------+---------------------------------------------------------------------------------------------+
  5| hive_db  | CREATE DATABASE `hive_db` LOCATION 'hdfs://172.21.16.47:4007/usr/hive/warehouse/hive_db.db' |
  6+----------+---------------------------------------------------------------------------------------------+

• 删除

  Plain Text 复制

  1DROP DATABASE [IF EXISTS] hive.hive_db;

  注意： 对于 Hive Database，必须先删除这个 Database 下的所有表后，才能删除 Database，否则会报错。这个操作会同步删除 Hive 中对应的 Database。

Table

• 创建

  PALO 支持在 Hive 中创建分区或非分区表。

  Plain Text 复制

  1-- Create unpartitioned hive table
  2CREATE TABLE unpartitioned_table (
  3  `col1` BOOLEAN COMMENT 'col1',
  4  `col2` INT COMMENT 'col2',
  5  `col3` BIGINT COMMENT 'col3',
  6  `col4` CHAR(10) COMMENT 'col4',
  7  `col5` FLOAT COMMENT 'col5',
  8  `col6` DOUBLE COMMENT 'col6',
  9  `col7` DECIMAL(9,4) COMMENT 'col7',
  10  `col8` VARCHAR(11) COMMENT 'col8',
  11  `col9` STRING COMMENT 'col9'
  12)  ENGINE=hive
  13PROPERTIES (
  14  'file_format'='parquet'
  15);
  16
  17-- Create partitioned hive table
  18-- The partition columns must be in table's column definition list
  19CREATE TABLE partition_table (
  20  `col1` BOOLEAN COMMENT 'col1',
  21  `col2` INT COMMENT 'col2',
  22  `col3` BIGINT COMMENT 'col3',
  23  `col4` DECIMAL(2,1) COMMENT 'col4',
  24  `pt1` VARCHAR COMMENT 'pt1',
  25  `pt2` VARCHAR COMMENT 'pt2'
  26)  ENGINE=hive
  27PARTITION BY LIST (pt1, pt2) ()
  28PROPERTIES (
  29  'file_format'='orc',
  30  'compression'='zlib'
  31);
  32
  33-- Create text format table(Since 2.1.7 & 3.0.3)
  34CREATE TABLE text_table (
  35    `id` INT,
  36    `name` STRING
  37) PROPERTIES (
  38    'file_format'='text',
  39    'compression'='gzip',
  40    'field.delim'='\t',
  41    'line.delim'='\n',
  42    'collection.delim'=';',
  43    'mapkey.delim'=':',
  44    'serialization.null.format'='\\N',
  45    'escape.delim'='\\'
  46);

  创建后，可以通过 SHOW CREATE TABLE 命令查看 Hive 的建表语句。

  注意，不同于 Hive 中的建表语句。在 PALO 中创建 Hive 分区表时，分区列也必须写到 Table 的 Schema 中。同时，分区列必须在所有 Schema 的最后，且顺序保持一致。

  注意：

  对于某些默认开启 ACID 事务特性的 Hive 集群，使用 PALO 建表后，表属性 transactional 会为 true。而 PALO 只支持部分 Hive 事务表的特性，因此可能会导致 PALO 创建的 Hive，PALO 本身无法读取的问题。因此，需要在建表的属性中，显式增加："transactional" = "false"，来创建非事务的 Hive 表：

  Plain Text 复制

  1CREATE TABLE non_acid_table(
  2  `col1` BOOLEAN COMMENT 'col1',
  3  `col2` INT COMMENT 'col2',
  4  `col3` BIGINT COMMENT 'col3'
  5)  ENGINE=hive
  6PROPERTIES (
  7  'transactional'='false',
  8);

• 删除

  可以通过 DROP TABLE 语句删除一个 Hive 表。当前删除表后，会同时删除数据，包括分区数据。

• 列类型

  在 PALO 中创建 Hive 表所使用的列类型，和 Hive 中的列类型对应关系如下

  PALO	Hive
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INT	INT
  BIGINT	BIGINT
  DATE	DATE
  DATETIME	TIMESTAMP
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  CHAR	CHAR
  VARCHAR	STRING
  STRING	STRING
  DECIMAL	DECIMAL
  ARRAY	ARRAY
  MAP	MAP
  STRUCT	STRUCT

  • 列类型只能为默认的 Nullable，不支持 NOT NULL。
  • Hive 3.0 支持设置默认值。如果需要设置默认值，则需要再 Catalog 属性中显示的添加 "hive.version" = "3.0.0"
  • 插入数据后，如果类型不能够兼容，例如 'abc' 插入到数值类型，则会转为 null 值后插入。

• 分区

  Hive 中的分区类型对应 PALO 中的 List 分区。因此，在 PALO 中 创建 Hive 分区表，需使用 List 分区的建表语句，但无需显式的枚举各个分区。在写入数据时，PALO 会根据数据的值，自动创建对应的 Hive 分区。

  支持创建单列或多列分区表。

• 文件格式

  • ORC（默认）
  • Parquet

  • Text（自 2.1.7 和 3.0.3 版本开始支持）

    Text 格式还支持以下表属性：

    • field.delim：列分隔符。默认 \1。
    • line.delim：行分隔符。默认 \n。
    • collection.delim：复杂类型中各元素之间的分隔符。默认 \2。
    • mapkey.delim：Map 类型的键值分割符。默认 \3
    • serialization.null.format：NULL 值的存储格式。默认 \N。
    • escape.delim：转移字符。默认 \。

• 压缩格式

  • Parquet：snappy（默认）、zstd、plain。（Plain 就是不采用压缩）
  • ORC：snappy、zlib（默认）、zstd、plain。（Plain 就是不采用压缩）
  • Text：gzip、defalte、bzip2、zstd、lz4、lzo、snappy、plain（默认）。（Plain 就是不采用压缩）

• 存储介质

  • HDFS
  • 对象存储

数据操作

可以通过 INSERT 语句将数据写入到 Hive 表中。

支持写入到由 PALO 创建的 Hive 表，或者 Hive 中已存在的且格式支持的表。

对于分区表，会根据数据，自动写入到对应分区，或者创建新的分区。

目前不支持指定分区写入。

INSERT

INSERT 操作会数据以追加的方式写入到目标表中。当前不支持指定分区写入。

1INSERT INTO hive_tbl values (val1, val2, val3, val4);
2INSERT INTO hive.hive_db.hive_tbl SELECT col1, col2 FROM internal.db1.tbl1;
3
4INSERT INTO hive_tbl(col1, col2) values (val1, val2);
5INSERT INTO hive_tbl(col1, col2, partition_col1, partition_col2) values (1, 2, "beijing", "2023-12-12");

INSERT OVERWRITE

INSERT OVERWRITE 会使用新的数据完全覆盖原有表中的数据。当前不支持指定分区写入。

1INSERT OVERWRITE TABLE VALUES(val1, val2, val3, val4)
2INSERT OVERWRITE TABLE hive.hive_db.hive_tbl(col1, col2) SELECT col1, col2 FROM internal.db1.tbl1;

INSERT OVERWRITE 的语义与 Hive 一致，有如下行为：

• 当目的表是分区表，而源表为空表时，操作不会产生任何影响。目的表数据无变化。
• 当目的表是非分区表，而源表是空表是，目的表会被清空。
• 当前不支持指定分区写入，因此 INSERT OVERWRITE 为根据源表中的数值，自动处理对应的目的表分区。如果目的表是分区表，则只会覆盖涉及到的分区，不涉及的分区，数据无变化。

CTAS(CREATE TABLE AS SELECT)

可以通过 CTAS(CREATE TABLE AS SELECT) 语句创建 Hive 表并写入数据：

1CREATE TABLE hive_ctas ENGINE=hive AS SELECT * FROM other_table;

CTAS 支持指定文件格式、分区方式等信息，如：

1CREATE TABLE hive_ctas ENGINE=hive
2PARTITION BY LIST (pt1, pt2) ()
3AS SELECT col1,pt1,pt2 FROM part_ctas_src WHERE col1>0;
4    
5CREATE TABLE hive.hive_db.hive_ctas (col1,col2,pt1) ENGINE=hive
6PARTITION BY LIST (pt1) ()
7PROPERTIES (
8    "file_format"="parquet",
9    "compression"="zstd"
10)
11AS SELECT col1,pt1 as col2,pt2 as pt1 FROM test_ctas.part_ctas_src WHERE col1>0;

事务机制

对 Hive 的写入操作会被放在一个单独的事务里，在事务提交前，数据对外不可见。只有当提交该事务后，表的相关操作才对其他人可见。

事务能保证操作的原子性，事务内的所有操作，要么全部成功，要么全部失败。

事务不能完全保证操作的隔离性，只能尽力而为，通过分离文件系统操作和 对 Hive Metastore 的元数据操作来尽量减少不一致的时间窗口。

比如在一个事务中，需要修改 Hive 表的多个分区。假设这个任务分成两批进行操作，在第一批操作已经完成、第二批操作还未完成时，第一批分区已经对外可见，外部可以读取到第一批分区，但读不到第二批分区。

在事务提交过程中出现任何异常，都会直接回退该事务，包括对 HDFS 文件的修改、以及对 Hive Metastore 元数据的修改，不需要用户做其他处理。

并发写入机制

当前 PALO 支持使用多个插入语句进行并发写入。不过需要注意的是，用户需要控制并发写入不产生可能冲突的情况。

因为普通非事务 Hive 表缺少完备的事务机制。通过上文介绍的 PALO 事务机制我们知道目前 PALO 中的实现只能是尽力而为地减少可能不一致的时间窗口，而无法保证真正的 ACID。因此在 PALO 中进行并发写入 Hive 表可能会导致数据一致性问题。

1. INSERT 并发操作

   INSERT 为数据追加操作，在并发执行 INSERT 时，不会产生冲突，操作会产生预期的结果。

2. INSERT OVERWRITE 并发操作

   如果使用 INSERT OVERWRITE 对同一表或分区并发写入，可能会导致数据丢失或损坏，结果可能是不确定的。

   一般有以下几种解决方案：

   • 对于分区表，可以将数据写入不同的分区，并发操作不同分区不会产生冲突。
   • 对于非分区表，可以同时执行 INSERT，而不使用 INSERT OVERWRITE，这样不会产生冲突的问题。
   • 对于可能产生冲突的操作，需要用户在业务侧控制同一时间只有一个写入在进行。

HDFS 文件操作

在 HDFS 上的 Hive 表数据通常会先写入到临时目录，然后通过 rename 等文件系统操作进行最终的文件提交。这里我们详细介绍不同数据操作中，HDFS 上文件的具体操作。

数据的临时目录格式为：/tmp/.doris_staging/<username>/<uuid>

写入的数据文件名称格式为：<query-id>_<uuid>-<index>.<compress-type>.<file-type>

下面举例说明各种情况下的文件操作。

1. 非分区表

   • Append（追加写入）

     • 目标表目录：hdfs://ns/usr/hive/warehouse/example.db/table1
     • 临时文件：hdfs://ns/tmp/.doris_staging/root/f02247cb662846038baae272af5eeb05/b35fdbcea3a4e39-86d1f36987ef1492_7e3985bf-9de9-4fc7-b84e-adf11aa08756-0.orc

     提交阶段会把所有临时文件移动到目标表目录下。

   • Overwrite（覆盖写）

     • 目标表目录：hdfs://ns/usr/hive/warehouse/example.db/table1
     • 临时文件：hdfs://ns/tmp/.doris_staging/root/f02247cb662846038baae272af5eeb05/b35fdbcea3a4e39-86d1f36987ef1492_7e3985bf-9de9-4fc7-b84e-adf11aa08756-0.orc

     提交阶段：

     1. 目标表目录重命名为目标表临时目录：hdfs://ns/usr/hive/warehouse/example.db/_temp_b35fdbcea3a4e39-86d1f36987ef1492_table1
     2. 临时目录重命名为目标表目录。
     3. 删除目标表临时目录。

2. 分区表

   • Add（添加到新分区）

     • 目标表目录：hdfs://ns/usr/hive/warehouse/example.db/table2/part_col=2024-01-01
     • 临时文件：hdfs://ns/tmp/.doris_staging/root/a7eac7505d7a42fdb06cb9ef1ea3e912/par1=a/d678a74d232345e0-b659e2fb58e86ffd_549ad677-ee75-4fa1-b8a6-3e821e1dae61-0.orc

     提交阶段，会将临时目录重命名为目标表目录

   • Append（写入数据到已存在的分区）

     • 目标表目录：hdfs://ns/usr/hive/warehouse/example.db/table2/part_col=2024-01-01
     • 临时文件：hdfs://ns/tmp/.doris_staging/root/a7eac7505d7a42fdb06cb9ef1ea3e912/par1=a/d678a74d232345e0-b659e2fb58e86ffd_549ad677-ee75-4fa1-b8a6-3e821e1dae61-0.orc

     提交阶段，会将临时目录下的文件，移动到目标表目录下。

   • Overwrite（覆盖已有分区）

     • 目标表目录：hdfs://ns/usr/hive/warehouse/example.db/table2/part_col=2024-01-01
     • 临时文件：hdfs://ns/tmp/.doris_staging/root/a7eac7505d7a42fdb06cb9ef1ea3e912/par1=a/d678a74d232345e0-b659e2fb58e86ffd_549ad677-ee75-4fa1-b8a6-3e821e1dae61-0.orc

     提交阶段：

     1. 目标表分区目录重命名为目标表临时分区目录：hdfs://ns/usr/hive/warehouse/example.db/table2/_temp_d678a74d232345e0-b659e2fb58e86ffd_part_col=2024-01-01
     2. 临时分区目录重命名为目标表分区目录。
     3. 删除目标表临时分区目。

相关参数

BE