作用:Catalog 为了保证不同数据源的访问方式的统一而建立

Catalog 提供了一个统一的API,用于管理元数据,并使其可以从 Table API 和 SQL 查询语句中来访问。
前面用到Connector其实就是在使用Catalog

同一管理不同数据库的表,相当于给数据库建一层目录:
catalog.db.table:
虽然不同的数据源要用不同的catalog来访问,但是catalog起的名字可以相同,这样就可以统一管理

-目录
-kafka
-表1, …..
-mysql
-……
-hbase
-clickhouse
-hive

其实我们所有API其实都有默认的Catalog(GenericInMemoryCatalog) https://www.yuque.com/cheng-rtxtv/kb/doww38#hkKAo
image.png

Catalog类型

GenericInMemoryCatalog(默认)

GenericInMemoryCatalog 是基于内存实现的 Catalog,所有元数据只在 session 的生命周期内可用。

JdbcCatalog

JdbcCatalog 使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。PostgresCatalog 是当前实现的唯一一种 JDBC Catalog。

HiveCatalog

HiveCatalog 有两个用途:作为原生 Flink 元数据的持久化存储,以及作为读写现有 Hive 元数据的接口。 Flink 的 Hive 文档 提供了有关设置 HiveCatalog 以及访问现有 Hive 元数据的详细信息。

HiveCatalog

1、导入需要的依赖

  1. <dependency>
  2. <groupId>org.apache.flink</groupId>
  3. <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
  4. <version>${flink.version}</version>
  5. </dependency>
  6. <!-- Hive Dependency -->
  7. <dependency>
  8. <groupId>org.apache.hive</groupId>
  9. <artifactId>hive-exec</artifactId>
  10. <version>3.1.2</version>
  11. </dependency>

2、在hadoop162启动hive元数据,

flink连接hive要求一定要开启元数据服务!
原理:先去找元数据服务代理,再去找装载元数据的数据库(mysql)
image.png

nohup hive —service metastore >/dev/null 2>&1 &

3连接Hive

| StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

String name = “myhive”; // Catalog 名字
_String defaultDatabase = “flink_test”;
// 默认数据库
String hiveConfDir = “c:/conf”; // hive配置文件的目录.** 需要把hive-site.xml添加到该目录(测试的时候是windows本地的目录)_**

// 1. 创建HiveCatalog
_HiveCatalog hive = new HiveCatalog**
(name, defaultDatabase, hiveConfDir)**;
// 2. 注册HiveCatalogtEnv.registerCatalog(name, hive);
// 3. 把 HiveCatalog: myhive 作为当前session的catalog
_tEnv.useCatalog**
(name);
tEnv.useDatabase
(“flink_test”);
tEnv.sqlQuery
(“select * from stu”).execute().print()_**; | | —- |

_