大数据数据湖数据仓库 6 Hive引擎集成Apache Paimon

更多Paimon数据湖内容请关注：https://edu.51cto.com/course/35051.html

在实际工作中，我们通查会使用Flink计算引擎去读写Paimon，但是在批处理场景中，更多的是使用Hive去读写Paimon，这样操作起来更加方便。

前面我们在Flink代码里面，借助于Hive Catalog，实现了在Flink中创建Paimon表，写入数据，并且把paimon的元数据信息保存在Hive Metastore里面，这样创建的表是可以被Hive识别并且操作的。

但是最直接的肯定是在Hive中直接创建Paimon类型的表，并且读写数据。

Paimon目前可以支持Hive 3.1, 2.3, 2.2, 2.1 and 2.1-cdh-6.3这些版本的操作。

但是需要注意，如果Hive的执行引擎使用的是Tez，那么只能读取Paimon，无法向Paimon中写入数据。如果Hive的执行引擎使用的是MR，那么读写都是支持的。

在Hive中配置Paimon依赖

想要在Hive中操作Paimon，首先需要在Hive中配置Paimon的依赖，此时我们需要用到一个jar包：paimon-hive-connector。我们目前使用的Hive是3.1.2版本的，所以需要下载对应版本的paimon-hive-connector jar包。

https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-hive-connector-3.1/0.5.0-incubating/paimon-hive-connector-3.1-0.5.0-incubating.jar

将这个jar包上传到bigdata04机器（hive客户端机器）的hive安装目录中：

[root@bigdata04 ~]# cd /data/soft/apache-hive-3.1.2-bin/

[root@bigdata04 apache-hive-3.1.2-bin]# mkdir auxlib

[root@bigdata04 apache-hive-3.1.2-bin]# cd auxlib/

[root@bigdata04 auxlib]# ll

total 34128

-rw-r--r--. 1 root root 34945743 Sep 13 2023 paimon-hive-connector-3.1-0.5.0-incubating.jar

注意：需要在hive安装目录中创建auxlib目录，然后把jar包上传到这个目录中，这样会被自动加载。

如果我们在操作Hive的时候使用的是beeline客户端，那么在Hive中配置好Paimon的环境之后，需要重启HiveServer2服务。

在Hive中读写Paimon表

咱们之前在Flink引擎代码中使用Hive Catalog的时候创建了一个表：p_h_t1，这个表的元数据在Hive Metastore也有存储，之前我们其实也能查看到，只是在hive中查询这个表中数据的时候报错了，其实就是因为缺少paimon-hive-connector这个jar包，现在我们再查询就可以了。

在这里我们使用Hive的beeline客户端。注意：需要先启动HiveServer2服务。

[root@bigdata04 ~]# cd /data/soft/apache-hive-3.1.2-bin/

[root@bigdata04 apache-hive-3.1.2-bin]# bin/hiveserver2

查看hive中目前都有哪些表：

[root@bigdata04 apache-hive-3.1.2-bin]# bin/beeline -u jdbc:hive2://localhost:10000 -n root

Connecting to jdbc:hive2://localhost:10000

Connected to: Apache Hive (version 3.1.2)

Driver: Hive JDBC (version 3.1.2)

Transaction isolation: TRANSACTION_REPEATABLE_READ

Beeline version 3.1.2 by Apache Hive

0: jdbc:hive2://localhost:10000> SHOW TABLES;

+--------------------+

| tab_name |

+--------------------+

| flink_stu |

| orders |

| p_h_t1 |

| p_h_par |

| s1 |

| student_favors |

| student_favors_2 |

| student_score |

| student_score_bak |

| t1 |

+--------------------+

9 rows selected (1.663 seconds)

此时可以看到之前通过Hive Catalog写入的表：p_h_t1。

查询这个表中的数据：

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t1;

+--------------+-------------+

| p_h_t1.name | p_h_t1.age |

+--------------+-------------+

| jack | 18 |

| tom | 20 |

+--------------+-------------+

2 rows selected (5.853 seconds)

此时就可以正常查询了。

接着我们尝试在Hive中向这个Paimon表中插入一条数据：

0: jdbc:hive2://localhost:10000> INSERT INTO p_h_t1(name,age) VALUES('jessic',19);

重新查询这个表中的最新数据：

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t1;

+--------------+-------------+

| p_h_t1.name | p_h_t1.age |

+--------------+-------------+

| jack | 18 |

| jessic | 19 |

| tom | 20 |

+--------------+-------------+

3 rows selected (0.737 seconds)

在通过Hive进行查询的时候，默认查询的是表中最新快照的数据，我们也可以通过时间旅行这个特性来控制查询之前的数据。

举个例子，查询指定快照版本中的数据：

0: jdbc:hive2://localhost:10000> SET paimon.scan.snapshot-id=1;

No rows affected (0.011 seconds)

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t1;

+--------------+-------------+

| p_h_t1.name | p_h_t1.age |

+--------------+-------------+

| jack | 18 |

| tom | 20 |

+--------------+-------------+

2 rows selected (0.752 seconds)

0: jdbc:hive2://localhost:10000> SET paimon.scan.snapshot-id=2;

No rows affected (0.009 seconds)

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t1;

+--------------+-------------+

| p_h_t1.name | p_h_t1.age |

+--------------+-------------+

| jack | 18 |

| jessic | 19 |

| tom | 20 |

+--------------+-------------+

3 rows selected (0.692 seconds)

这样就可以实现查询历史数据的查询了。

在Hive中创建Paimon表

前面我们操作的p_h_t1这个表其实是借助于Flink引擎创建的。下面我们来看一下在Hive中如何创建Piamon表：

0: jdbc:hive2://localhost:10000> SET hive.metastore.warehouse.dir=hdfs://bigdata01:9000/paimon;

0: jdbc:hive2://localhost:10000> CREATE TABLE IF NOT EXISTS p_h_t2(

name STRING,

age INT,

PRIMARY KEY (name) NOT ENFORCED

)STORED BY 'org.apache.paimon.hive.PaimonStorageHandler';

这样表就创建好了，下面我们可以在Hive中测试一下读写数据：

0: jdbc:hive2://localhost:10000> INSERT INTO p_h_t2(name,age) VALUES('tom',20);

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t2;

Error: java.io.IOException: java.lang.RuntimeException: Fails to read snapshot from path hdfs://bigdata01:9000/paimon/default.db/p_h_t2/snapshot/snapshot-2 (state=,code=0)

注意：此时查询报错是因为找不到snapshot-2这份快照数据，目前这个表中只添加了一次数据，所以只有snapshot-1。

那为什么会查找snapshot-2呢？

因为我们前面在这个会话中设置了SET paimon.scan.snapshot-id=2;，这个配置在当前会话有效。

正常情况下我们在hive中执行SET paimon.scan.snapshot-id=null;其实就可以了：

0: jdbc:hive2://localhost:10000> SET paimon.scan.snapshot-id=null;

No rows affected (0.008 seconds)

0: jdbc:hive2://localhost:10000> SET paimon.scan.snapshot-id;

+-------------------------------+

| set |

+-------------------------------+

| paimon.scan.snapshot-id=null |

+-------------------------------+

1 row selected (0.009 seconds)

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t2;

Error: java.io.IOException: java.lang.RuntimeException: Fails to read snapshot from path hdfs://bigdata01:9000/paimon/default.db/p_h_t2/snapshot/snapshot-2 (state=,code=0)

但是发现他还是会找snapshot-2。

我们尝试重新开启一个新的会话查询也不行，就算重启hiveserver2也还是不行。

后来发现这可能是一个bug，当我们在hive会话中设置了paimon.scan.snapshot-id=2，那么之后创建的表默认就只会查询snapshot-2了，那也就意味着建表的时候会把这个参数带过去。

为了验证这个猜想，我们在flink代码中查询这个Paimon表的详细建表语句，不要在hive命令行中查看（在Hive命令行中看不到详细的参数信息）。

创建package：tech.xuwei.paimon.hivepaimon 创建object：FlinkSQLReadFromPaimo

完整代码如下：

package tech.xuwei.paimon.cdcingestion

import org.apache.flink.api.common.RuntimeExecutionMode

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**

* 使用FlinkSQL从Paimon表中读取数据

* Created by xuwei

object FlinkSQLReadFromPaimon {

def main(args: Array[String]): Unit = {

//创建执行环境

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setRuntimeMode(RuntimeExecutionMode.STREAMING)

val tEnv = StreamTableEnvironment.create(env)

//创建Paimon类型的Catalog

tEnv.executeSql(

"""

|CREATE CATALOG paimon_catalog WITH(

| 'type'='paimon',

| 'warehouse'='hdfs://bigdata01:9000/paimon'

|""".stripMargin)

tEnv.executeSql("USE CATALOG paimon_catalog")

//读取Paimon表中的数据，并且打印输出结果

tEnv.executeSql(

"""

|SHOW CREATE TABLE p_h_t2;

|""".stripMargin)

.print()

}

在idea中执行代码，查看结果：

CREATE TABLE `paimon_catalog`.`default`.`p_h_t2` (

`name` VARCHAR(2147483647),

`age` INT

) WITH (

'path' = 'hdfs://bigdata01:9000/paimon/default.db/p_h_t2',

'totalSize' = '0',

'numRows' = '0',

'rawDataSize' = '0',

'scan.snapshot-id' = '2',

'COLUMN_STATS_ACCURATE' = '{"BASIC_STATS":"true","COLUMN_STATS":{"age":"true","name":"true"}}',

'numFiles' = '0',

'bucketing_version' = '2',

'storage_handler' = 'org.apache.paimon.hive.PaimonStorageHandler'

)

在这里发现建表语句中有一个参数：'scan.snapshot-id' = '2'，所以它默认会读取第2个快照。

想要解决这个问题，有两个办法。

1：在hive中删除这个表，然后执行SET paimon.scan.snapshot-id=null;，再创建这个表就行了。2：如果不想删除这个表，可以在Flink代码中修改这个表，移除scan.snapshot-id属性即可，这个功能我们之前讲过。

第一种办法简单粗暴，不再演示，我们来看一下第二种办法：

创建object：FlinkSQLAlterPaimonTable

完整代码如下：

package tech.xuwei.paimon.hivepaimon

import org.apache.flink.api.common.RuntimeExecutionMode

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**

* 修改Paimon表属性

* Created by xuwei

object FlinkSQLAlterPaimonTable {

def main(args: Array[String]): Unit = {

//创建执行环境

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setRuntimeMode(RuntimeExecutionMode.STREAMING)

val tEnv = StreamTableEnvironment.create(env)

//创建Paimon类型的Catalog

tEnv.executeSql(

"""

|CREATE CATALOG paimon_catalog WITH(

| 'type'='paimon',

| 'warehouse'='hdfs://bigdata01:9000/paimon'

|""".stripMargin)

tEnv.executeSql("USE CATALOG paimon_catalog")

//移除表中的scan.snapshot-id属性

tEnv.executeSql(

"""

|ALTER TABLE p_h_t2 RESET ('scan.snapshot-id')

|""".stripMargin)

//查看最新的表属性信息

tEnv.executeSql(

"""

|SHOW CREATE TABLE p_h_t2

|""".stripMargin)

.print()

}

执行此代码，可以看到如下结果：

CREATE TABLE `paimon_catalog`.`default`.`p_h_t2` (

`name` VARCHAR(2147483647),

`age` INT

) WITH (

'path' = 'hdfs://bigdata01:9000/paimon/default.db/p_h_t2',

'totalSize' = '0',

'numRows' = '0',

'rawDataSize' = '0',

'COLUMN_STATS_ACCURATE' = '{"BASIC_STATS":"true","COLUMN_STATS":{"age":"true","name":"true"}}',

'numFiles' = '0',

'bucketing_version' = '2',

'storage_handler' = 'org.apache.paimon.hive.PaimonStorageHandler'

)

此时表中就没有scan.snapshot-id属性了。

这个时候我们再回到hive命令行中查询这个表：

0: jdbc:hive2://localhost:10000> SELECT * FROM p_h_t2;

+--------------+-------------+

| p_h_t2.name | p_h_t2.age |

+--------------+-------------+

| tom | 20 |

+--------------+-------------+

1 row selected (0.46 seconds)

这样就可以正常查询了。

注意：如果此时我们在hive中删除这个表，那么对应的paimon中这个表也会被删除。

0: jdbc:hive2://localhost:10000> drop table p_h_t2;

No rows affected (0.33 seconds)

到hdfs中确认一下这个paimon表是否存在：

[root@bigdata04 ~]# hdfs dfs -ls /paimon/default.db/p_h_t2

ls: `/paimon/default.db/p_h_t2': No such file or directory

这样就说明paimon中这个表不存在了。

不过在hdfs中会多一个这个目录，这属于一个临时目录，没什么影响，可以手工删除，不处理也没影响。

[root@bigdata04 ~]# hdfs dfs -ls /paimon/default.db/_tmp.p_h_t2

针对Paimon中已经存在的表，我们想要在hive中进行访问，应该如何实现呢？

此时可以借助于Hive的外部表特性来实现。相当于是在hive中创建一个外部表，通过location指向paimon表的hdfs路径即可。

我们使用前面cdc数据采集中创建的Paimon表：cdc_chinese_code，在hive中创建一个外部表映射到这个表：

0: jdbc:hive2://localhost:10000> CREATE EXTERNAL TABLE p_h_external

STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'

LOCATION 'hdfs://bigdata01:9000/paimon/default.db/cdc_chinese_code';

然后就可以在hive中查询这个表了：

0: jdbc:hive2://localhost:10000> select * from p_h_external;

+------------------+--------------------+

| p_h_external.id | p_h_external.name |

+------------------+--------------------+

| 1 | 张三 |

| 2 | 李四 |

+------------------+--------------------+

此时如果我们在hive中删除这个外部表，不会影响paimon中的cdc_chinese_code表。

0: jdbc:hive2://localhost:10000> drop table p_h_external;

到hdfs中确认一下，cdc_chinese_code这个paimon表还是存在的：

[root@bigdata04 ~]# hdfs dfs -ls /paimon/default.db/cdc_chinese_code

Found 4 items

drwxr-xr-x - root supergroup 0 2029-02-27 11:32 /paimon/default.db/cdc_chinese_code/bucket-0

drwxr-xr-x - root supergroup 0 2029-02-27 11:32 /paimon/default.db/cdc_chinese_code/manifest

drwxr-xr-x - root supergroup 0 2029-02-27 11:26 /paimon/default.db/cdc_chinese_code/schema

drwxr-xr-x - root supergroup 0 2029-02-27 11:32 /paimon/default.db/cdc_chinese_code/snapshot

更多Paimon数据湖内容请关注：https://edu.51cto.com/course/35051.html

参考阅读

评论可见，请评论后查看内容，谢谢！！！

您阅读本篇文章共花了：

金钥匙

大数据数据湖数据仓库 6 Hive引擎集成Apache Paimon

笔记大数据Hadoop之——部署hadoop+hive+Mysql环境（Linux）

hive hadoop 数据仓库大数据今天又get到一个小技能，collect

发表评论取消回复

金钥匙

大数据 数据湖 数据仓库 6 Hive引擎集成Apache Paimon

笔记 大数据Hadoop之——部署hadoop+hive+Mysql环境（Linux）

hive hadoop 数据仓库 大数据 今天又get到一个小技能，collect

相关文章

发表评论取消回复

大数据数据湖数据仓库 6 Hive引擎集成Apache Paimon

笔记大数据Hadoop之——部署hadoop+hive+Mysql环境（Linux）

hive hadoop 数据仓库大数据今天又get到一个小技能，collect