万能的Hive Metastore能存哪些类型的表？

Posted 2022-11-12 咬定青松

本文首发微信公众号：码上观世界

由于Hive Metastore在Hadoop生态中的独特地位，Hive Metastore也顺理成章成为实时大数据领域中元数据存储的”兵家必争之地“。目前，主要引擎Spark、Flink、Iceberg、Trino等都集成了Hive Metastore。

那么Hive Metastore这个万能元数据大熔炉到底能存储哪些类型的数据表呢？简单概括为以下几种：

• 普通的Hive表；

• Hive外表；

• Flink映射表；

• Iceberg表；

如何存储Hive表？

先看看Hive Metastore如何定义和存储表的。Hive Metastore表分为managed表和external表，两者的区别是：

A managed table is one for which the definition is primarily managed in Hive's metastore, and for whose data storage Hive is responsible. An external table is one whose definition is managed in some external catalog, and whose data Hive does not own (i.e. it will not be deleted when the table is dropped).

简单说就是managed表的元数据和数据都由Hive Metastore负责管理，而Hive Metastore只管理external表的元数据，不管理它的数据，定义外表，只需要加上关键字external即可：

CREATE EXTERNAL TABLE my_external_table (a string, b string)
LOCATION '/user/data';

但hive storage handler特性又引入了native表和non-native表，前者是创建表的时候不需要通过STORED BY指定额外处理类即可识别外部数据格式，后者则需要明确指定额外处理类才能识别外部数据格式的表。

这样，native表和non-native表以及前文中提到的managed表和external表，组成4种组合表：

• managed native: what you get by default with CREATE TABLE

• external native: what you get with CREATE EXTERNAL TABLE when no STORED BY clause is specified

• managed non-native: what you get with CREATE TABLE when a STORED BY clause is specified; Hive stores the definition in its metastore, but does not create any files itself; instead, it calls the storage handler with a request to create a corresponding object structure

• external non-native: what you get with CREATE EXTERNAL TABLE when a STORED BY clause is specified; Hive registers the definition in its metastore and calls the storage handler to check that it matches the primary definition in the other system

HMS存储表相关的元数据表主要涉及下面几个：

但是在HMS TBLS中，只会根据字段属性TBL_TYPE来区分内外表，TBL_TYPE取值为[MANAGED_TABLE,EXTERNAL_TABLE]。

内外表如何区分存储？

存储一个普通的 managed native表，主要涉及到的表为TBLS、TBL_PARAMS和COLUMNS_V2，其中COLUMNS_V2存储表的字段Schema信息。而存储一个 external non-native表，主要涉及到的表为TBLS、TBL_PARAMS、COLUMNS_V2和SERDE_PARAMS。举个常见的hive 访问hbase外表的实例：

create external table tb_user(id int,name string, age int) 
STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 
WITH 
SERDEPROPERTIES ("hbase.columns.mapping" = ":key,info:name,info:age") 
TBLPROPERTIES("hbase.table.name" = "user");

这是一个external non-native表，表的schema信息不是存储在COLUMNS_V2，而是将其映射关系存储在SERDE_PARAMS中：

SERDE_ID(TABLE_ID)	PARAM_KEY	PARAM_VALUE
668	hbase.columns.mapping	:key,info:name,info:age

参数信息存储在TABLE_PARMAS：

EXTERNAL表明了这是一个外表，当create表没有指定external关键字时，它是一个managed_table。

Flink映射表如何存储？

Flink Connector像"八爪鱼的触角"一样，能够延伸到多种数据源，它能够将不同数据源的表映射到自身管理的Catalog中。比如下面将mysql表table_params映射到Flink GeneralMemoryCatalog中：

-- register a MySQL table 'table_params' in Flink SQL
-- use hive catalog
CREATE TABLE flink_table (
    param_key  varchar(255),
    param_value varchar(255)
) WITH(
   'connector' = 'jdbc'
   'username'='root',
   'password'='***',
   'url' = 'jdbc:mysql://10.201.0.205:3306/hivemetastore_db_dev_212',
   'table-name' = 'table_params'
)

jdbc映射表在HMS中被当做managed_table来看待的，但是表的schema信息不是存储在COLUMNS_V2，而是将其映射关系存储在TABLE_PARAMS中：

因为映射mysql表需要建立对mysql server的连接，故需要指定url和相应的连接信息，同时保证users表的存在。同样，Iceberg表也可以映射到Flink GeneralMemoryCatalog中。

CREATE TABLE flink_table (
    id   BIGINT,
    data STRING
) WITH (
    'connector'='iceberg',
    'catalog-name'='hive_prod',
    'catalog-database'='test_database',
    'catalog-table'='hive_iceberg_table_15',
    'uri'='thrift://localhost:9083',
    'warehouse'='hdfs://nn:8020/path/to/warehouse'
);

这个示例在Flink GeneralMemoryCatalog中创建了表flink_table，HMS的后端表库和表分别通过catalog-database和catalog-table指定。当对应的库和表不存在时，Flink会自动创建它们。

照理说，通过flink connector创建的表属于映射表，但是Iceberg表在HMS中被当做external_table来看待了，TABLE_PARAMS中存储的内容如下：

可见，Iceberg表元信息详情通过metadata_location指定，schema的版本记录等信息就存在其指向的文件中，而表的最新schema信息存储在COLUMNS_V2中。

Iceberg表有几种创建方式？

方式一：以映射表的方式来创建：

CREATE TABLE flink_table (
    id   BIGINT,
    data STRING
) WITH (
    'connector'='iceberg',
    ......
）

但是这种方式只能在Flink GeneralMemoryCatalog中创建，而其他类型的映射表，比如JDBC映射表，如果要持久化到Hive，是可以且必须使用Flink HiveCatalog。

方式二：通过FlinkCatalog创建Iceberg表

FlinkCatalog是Iceberg对Flink Catalog接口实现，支持Iceberg的Hive Catalog和Hadoop Catalog，下面是一个使用示例：

CatalogLoader catalogLoader=CatalogLoader.hive("hive", hiveConf, properties);
FlinkCatalog flinkCatalog=new FlinkCatalog("test_catalog_name","default", Namespace.empty(),catalogLoader,false);
flinkCatalog.open();
tableEnvironment.registerCatalog(flinkCatalog.getName(),flinkCatalog);
tableEnvironment.useCatalog(flinkCatalog.getName());
String tblName="test_iceberg_table_15";
tableEnvironment.executeSql(String.format("drop table if exists %s",tblName));
String sql= String.format("CREATE TABLE %s(" +
        "      id BIGINT COMMENT 'unique id'," +
        "      data STRING" +
        "     ),tblName);
tableEnvironment.executeSql(sql);

方式三：使用Flink SQL创建表：

CREATE CATALOG hive_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://localhost:9083',
  'clients'='5',
  'property-version'='1',
  'warehouse'='hdfs://nn:8020/warehouse/path'
);


CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING
);

这种方式实际上，通过Iceberg实现的FlinkCatalog来创建表，FlinkCatalog即上述Iceberg对Flink Catalog接口实现。

方式四：通过Hive Cli SQL

SET iceberg.catalog.another_hive.type=hive;
SET iceberg.catalog.another_hive.uri=thrift://master:9083;
SET iceberg.catalog.another_hive.clients=10;
SET iceberg.catalog.another_hive.warehouse=hdfs://master:9000/user/hive/warehouse/
create database if not exists test_db;
CREATE TABLE test_db.test_tbl (
  id bigint, name string
) PARTITIONED BY (
  dept string
) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
TBLPROPERTIES ('iceberg.catalog'='hive');

这种方式实际是使用了Hive的传统优势：通过storage handler实现了外表创建，按照前面对表的类型划分，就是 managed non-native。

参考：

https://iceberg.apache.org/docs/0.13.2/hive/

https://cwiki.apache.org/confluence/display/Hive/Managed+vs.+External+Tables

https://cwiki.apache.org/confluence/display/Hive/StorageHandlers

https://iceberg.apache.org/docs/0.13.2/hive/

https://hbase.apache.org/book.html#quickstart