4 Paimon数据湖之Hive Catalog的使用

2023-11-10 12:52

本文主要是介绍4 Paimon数据湖之Hive Catalog的使用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

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

Paimon提供了两种类型的Catalog:Filesystem CatalogHive Catalog

  • Filesystem Catalog:会把元数据信息存储到文件系统里面。
  • Hive Catalog:则会把元数据信息存储到Hive的Metastore里面,这样就可以直接在Hive中访问Paimon表了。注意:此时也会同时在文件系统中存储一份元数据信息,相当于元数据会存储两份,这个大家需要特别注意一下。

还有就是我们在使用Hive Catalog的时候,Paimon中的数据库名称、表名称,以及字段名称都要小写,因为这些数据存储到Hive Metastore的时候,会统一存储为小写。

下面我们来具体演示一下Paimon如何使用Hive Catalog来存储元数据。

在Flink中操作Paimon的时候想要使用Hive Catalog,需要依赖于Flink Hive connector,以及hive-execflink-table-api-scala-bridge

flink-table-api-scala-bridge这个依赖我们之前已经添加过了,所以只需要添加另外两个即可:

<!-- flink-hive-connector -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-hive_2.12</artifactId><version>1.15.0</version><!--<scope>provided</scope>-->
</dependency>
<dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>3.1.2</version><exclusions><exclusion><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId></exclusion></exclusions><!--<scope>provided</scope>-->
</dependency>

创建package:tech.xuwei.paimon.catalog
创建object:PaimonHiveCatalog

代码如下:

package tech.xuwei.paimon.catalogimport org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment/*** Paimon使用Hive Catalog* Created by xuwei*/
object PaimonHiveCatalog {def main(args: Array[String]): Unit = {//创建执行环境val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setRuntimeMode(RuntimeExecutionMode.STREAMING)val tEnv = StreamTableEnvironment.create(env)//创建Paimon类型的Catalog--使用Hive CatalogtEnv.executeSql("""|CREATE CATALOG paimon_hive_catalog WITH(|    'type'='paimon',|    'metastore' = 'hive',|    'uri' = 'thrift://bigdata04:9083',|    'warehouse'='hdfs://bigdata01:9000/paimon'|)|""".stripMargin)tEnv.executeSql("USE CATALOG paimon_hive_catalog")//创建Paimon表tEnv.executeSql("""|CREATE TABLE IF NOT EXISTS p_h_t1(|    name STRING,|    age INT,|    PRIMARY KEY (name) NOT ENFORCED|)|""".stripMargin)//向表中插入数据tEnv.executeSql("""|INSERT INTO p_h_t1(name,age) VALUES('jack',18),('tom',20)|""".stripMargin)}}

接下来到bigdata04节点上启动hive的metastore服务。

[root@bigdata04 ~]# cd /data/soft/apache-hive-3.1.2-bin/
[root@bigdata04 apache-hive-3.1.2-bin]# nohup bin/hive --service metastore -p 9083 2>&1 >/dev/null &

然后运行代码PaimonHiveCatalog

代码运行之后可以到先到hdfs中确认一下是否能看到元数据信息:

[root@bigdata04 ~]# hdfs dfs -cat /paimon/default.db/p_h_t1/schema/schema-0
{"id" : 0,"fields" : [ {"id" : 0,"name" : "name","type" : "STRING NOT NULL"}, {"id" : 1,"name" : "age","type" : "INT"} ],"highestFieldId" : 1,"partitionKeys" : [ ],"primaryKeys" : [ "name" ],"options" : { }

可以发现,在hdfs中依然是可以看到的,因为我们前面说了,使用hive catalog时也会同时在hdfs中存储一份元数据。

最后我们到hive中确认一下:
注意:由于目前bigdata04节点的环境变量中有HADOOP_CLASSPATH,所以直接使用hive客户端会看到很多日志信息,所以建议使用hive的beeline客户端。
此时需要先启动hiveserver2服务。

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

使用beeline客户端进行连接

[root@bigdata04 apache-hive-3.1.2-bin]# bin/beeline -u  jdbc:hive2://localhost:10000 -n root
0: jdbc:hive2://localhost:10000> show tables;
+--------------------+
|      tab_name      |
+--------------------+
| flink_stu          |
| orders             |
| p_h_t1             |
| s1                 |
| student_favors     |
| student_favors_2   |
| student_score      |
| student_score_bak  |
| t1                 |
+--------------------+
9 rows selected (1.727 seconds)
0: jdbc:hive2://localhost:10000> select * from p_h_t1;
Error: Error while compiling statement: FAILED: RuntimeException java.lang.ClassNotFoundException: org.apache.paimon.hive.mapred.PaimonInputFormat (state=42000,code=40000)

此时是可以在hive中查看到p_h_t1这个表的,但是在操作这个表的时候会报错,提示缺少依赖,现在报这个错是正常的,等后面我们会有一个单独的小节来讲Paimon和Hive引擎的集成。
目前通过hive catalog可以将paimon的元数据同时存储到hive的metastore中,但是还无法在hive中操作paimon的表,其实主要是因为缺少一个依赖,在这大家先知道这个问题即可。

注意:如果我们此时操作的是分区表,那么分区信息默认是无法同步到Hive Metastore的。

也就是说默认情况下,Paimon不会将新创建的分区同步到Hive Metastore中。我们在Hive中只能看到一个未分区的普通表。

如果想解决这个问题,也很简单,只需要在paimon的表属性中设置metastore.partitioned-table=true即可。

下面开发一个案例:
创建object:PaimonHiveCatalogPartitionTable,基于PaimonHiveCatalog进行复制。

完整代码如下:

package tech.xuwei.paimon.catalogimport org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment/*** Paimon使用Hive Catalog* 操作分区表* Created by xuwei*/
object PaimonHiveCatalogPartitionTable {def main(args: Array[String]): Unit = {//创建执行环境val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setRuntimeMode(RuntimeExecutionMode.STREAMING)val tEnv = StreamTableEnvironment.create(env)//创建Paimon类型的Catalog--使用Hive CatalogtEnv.executeSql("""|CREATE CATALOG paimon_hive_catalog WITH(|    'type'='paimon',|    'metastore' = 'hive',|    'uri' = 'thrift://bigdata04:9083',|    'warehouse'='hdfs://bigdata01:9000/paimon'|)|""".stripMargin)tEnv.executeSql("USE CATALOG paimon_hive_catalog")//创建Paimon表tEnv.executeSql("""|CREATE TABLE IF NOT EXISTS p_h_par(|    id INT,|    name STRING,|    dt STRING,|    PRIMARY KEY (id, dt) NOT ENFORCED|) PARTITIONED BY (dt) WITH(|    'metastore.partitioned-table' = 'true'|)|""".stripMargin)//向表中插入数据tEnv.executeSql("""|INSERT INTO p_h_par(id,name,dt)|VALUES(1,'jack','20230101'),(2,'tom','20230102')|""".stripMargin)}}

在idea中执行代码。

然后到hive中进行验证,可以执行show partitions p_h_par;进行验证。

或者到hive metastore里面进行确认,查看mysql中的partitions表,这个表里面存储的是分区信息,如果能看到分区信息,就说明Paimon表的分区信息同步过来了。
在这里插入图片描述

这样就说明Paimon表的分区信息同步过来了。

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

这篇关于4 Paimon数据湖之Hive Catalog的使用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/382756

相关文章

SpringBoot分段处理List集合多线程批量插入数据方式

《SpringBoot分段处理List集合多线程批量插入数据方式》文章介绍如何处理大数据量List批量插入数据库的优化方案:通过拆分List并分配独立线程处理,结合Spring线程池与异步方法提升效率... 目录项目场景解决方案1.实体类2.Mapper3.spring容器注入线程池bejsan对象4.创建

PHP轻松处理千万行数据的方法详解

《PHP轻松处理千万行数据的方法详解》说到处理大数据集,PHP通常不是第一个想到的语言,但如果你曾经需要处理数百万行数据而不让服务器崩溃或内存耗尽,你就会知道PHP用对了工具有多强大,下面小编就... 目录问题的本质php 中的数据流处理:为什么必不可少生成器:内存高效的迭代方式流量控制:避免系统过载一次性

Python使用FastAPI实现大文件分片上传与断点续传功能

《Python使用FastAPI实现大文件分片上传与断点续传功能》大文件直传常遇到超时、网络抖动失败、失败后只能重传的问题,分片上传+断点续传可以把大文件拆成若干小块逐个上传,并在中断后从已完成分片继... 目录一、接口设计二、服务端实现(FastAPI)2.1 运行环境2.2 目录结构建议2.3 serv

C#实现千万数据秒级导入的代码

《C#实现千万数据秒级导入的代码》在实际开发中excel导入很常见,现代社会中很容易遇到大数据处理业务,所以本文我就给大家分享一下千万数据秒级导入怎么实现,文中有详细的代码示例供大家参考,需要的朋友可... 目录前言一、数据存储二、处理逻辑优化前代码处理逻辑优化后的代码总结前言在实际开发中excel导入很

Spring Security简介、使用与最佳实践

《SpringSecurity简介、使用与最佳实践》SpringSecurity是一个能够为基于Spring的企业应用系统提供声明式的安全访问控制解决方案的安全框架,本文给大家介绍SpringSec... 目录一、如何理解 Spring Security?—— 核心思想二、如何在 Java 项目中使用?——

springboot中使用okhttp3的小结

《springboot中使用okhttp3的小结》OkHttp3是一个JavaHTTP客户端,可以处理各种请求类型,比如GET、POST、PUT等,并且支持高效的HTTP连接池、请求和响应缓存、以及异... 在 Spring Boot 项目中使用 OkHttp3 进行 HTTP 请求是一个高效且流行的方式。

Java使用Javassist动态生成HelloWorld类

《Java使用Javassist动态生成HelloWorld类》Javassist是一个非常强大的字节码操作和定义库,它允许开发者在运行时创建新的类或者修改现有的类,本文将简单介绍如何使用Javass... 目录1. Javassist简介2. 环境准备3. 动态生成HelloWorld类3.1 创建CtC

使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

《使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解》本文详细介绍了如何使用Python通过ncmdump工具批量将.ncm音频转换为.mp3的步骤,包括安装、配置ffmpeg环... 目录1. 前言2. 安装 ncmdump3. 实现 .ncm 转 .mp34. 执行过程5. 执行结

Java使用jar命令配置服务器端口的完整指南

《Java使用jar命令配置服务器端口的完整指南》本文将详细介绍如何使用java-jar命令启动应用,并重点讲解如何配置服务器端口,同时提供一个实用的Web工具来简化这一过程,希望对大家有所帮助... 目录1. Java Jar文件简介1.1 什么是Jar文件1.2 创建可执行Jar文件2. 使用java

C#使用Spire.Doc for .NET实现HTML转Word的高效方案

《C#使用Spire.Docfor.NET实现HTML转Word的高效方案》在Web开发中,HTML内容的生成与处理是高频需求,然而,当用户需要将HTML页面或动态生成的HTML字符串转换为Wor... 目录引言一、html转Word的典型场景与挑战二、用 Spire.Doc 实现 HTML 转 Word1