flink|实践数据湖iceberg 第三十二课 DDL语句通过hive catalog持久化方法 hive|kafka|big|data|iceberg

系列文章目录实践数据湖iceberg 第一课入门
实践数据湖iceberg 第二课 iceberg基于hadoop的底层数据格式
实践数据湖iceberg 第三课在sqlclient中，以sql方式从kafka读数据到iceberg
实践数据湖iceberg 第四课在sqlclient中，以sql方式从kafka读数据到iceberg（升级版本到flink1.12.7）
实践数据湖iceberg 第五课 hive catalog特点
实践数据湖iceberg 第六课从kafka写入到iceberg失败问题解决
实践数据湖iceberg 第七课实时写入到iceberg
实践数据湖iceberg 第八课 hive与iceberg集成
实践数据湖iceberg 第九课合并小文件
实践数据湖iceberg 第十课快照删除
实践数据湖iceberg 第十一课测试分区表完整流程(造数、建表、合并、删快照)
实践数据湖iceberg 第十二课 catalog是什么
实践数据湖iceberg 第十三课 metadata比数据文件大很多倍的问题
实践数据湖iceberg 第十四课元数据合并(解决元数据随时间增加而元数据膨胀的问题)
实践数据湖iceberg 第十五课 spark安装与集成iceberg(jersey包冲突)
实践数据湖iceberg 第十六课通过spark3打开iceberg的认知之门
实践数据湖iceberg 第十七课 hadoop2.7,spark3 on yarn运行iceberg配置
实践数据湖iceberg 第十八课多种客户端与iceberg交互启动命令(常用命令)
实践数据湖iceberg 第十九课 flink count iceberg，无结果问题
实践数据湖iceberg 第二十课 flink + iceberg CDC场景(版本问题，测试失败)
实践数据湖iceberg 第二十一课 flink1.13.5 + iceberg0.131 CDC(测试成功INSERT,变更操作失败)
实践数据湖iceberg 第二十二课 flink1.13.5 + iceberg0.131 CDC(CRUD测试成功)
实践数据湖iceberg 第二十三课 flink-sql从checkpoint重启
实践数据湖iceberg 第二十四课 iceberg元数据详细解析
实践数据湖iceberg 第二十五课后台运行flink sql 增删改的效果
实践数据湖iceberg 第二十六课 checkpoint设置方法
实践数据湖iceberg 第二十七课 flink cdc 测试程序故障重启：能从上次checkpoint点继续工作
实践数据湖iceberg 第二十八课把公有仓库上不存在的包部署到本地仓库
实践数据湖iceberg 第二十九课如何优雅高效获取flink的jobId
实践数据湖iceberg 第三十课 mysql->iceberg，不同客户端有时区问题
实践数据湖iceberg 第三十一课使用github的flink-streaming-platform-web工具，管理flink任务流，测试cdc重启场景
实践数据湖iceberg 第三十二课 DDL语句通过hive catalog持久化方法
实践数据湖iceberg 更多的内容目录

文章目录

系列文章目录
前言
一、flink 客户端配置hive catalog
- 1. 具体配置
- 2. 重新进入客户端，观察catalog变化
- 3.创建表，观察catalog对表持久化效果
总结

前言问题： flink的sql-client上，创建表，只是当前session有用，退出回话，需要重新创建表。多人共享一个表，很麻烦，有什么办法？
解决方法：把建表的DDL操作，持久化到HIVE上，由hive来管理。如何实现呢？使用hive catalog，在hive catalog下创建表。所有表都是持久化的。
官网的说明： https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/table/hive/hive_catalog/
一、flink 客户端配置hive catalog 1. 具体配置 【flink|实践数据湖iceberg 第三十二课 DDL语句通过hive catalog持久化方法】$FLNK_HOME/conf 下增加 sql-client-defaults.yaml 配置文件。
以下是配置默认catalog和database方法：

[root@hadoop101 conf]# cat sql-client-defaults.yaml execution: planner: blink type: streaming current-catalog: myhive# set the HiveCatalog as the current catalog of the session current-database: mydatabasecatalogs: - name: myhive type: hive hive-conf-dir: /opt/module/hive/conf

先通过hive创建这个database:

hive (default)> create database mydatabase; OK Time taken: 1.079 seconds

2. 重新进入客户端，观察catalog变化进入sql-client

Flink SQL> show current catalog; +----------------------+ | current catalog name | +----------------------+ |myhive | +----------------------+ 1 row in setFlink SQL> show current database; +-----------------------+ | current database name | +-----------------------+ |mydatabase | +-----------------------+ 1 row in set

3.创建表，观察catalog对表持久化效果

CREATE TABLE IF NOT EXISTS KafkaTableSource ( `log` STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'test1', 'properties.bootstrap.servers' = 'hadoop101:9092,hadoop102:9092,hadoop103:9092', 'properties.group.id' = 'testGroup', 'scan.startup.mode' = 'earliest-offset', 'format' = 'raw' );

Flink SQL> CREATE TABLE IF NOT EXISTS KafkaTableSource ( >`log` STRING > ) WITH ( >'connector' = 'kafka', >'topic' = 'test1', >'properties.bootstrap.servers' = 'hadoop101:9092,hadoop102:9092,hadoop103:9092', >'properties.group.id' = 'testGroup', >'scan.startup.mode' = 'earliest-offset', >'format' = 'raw' > ); [INFO] Execute statement succeed.

退出，重新进入客户端