Catalog : 提供元数据信息，如 : 数据库、表、分区、视图、数据库、其他外部系统中存储的函数和信息

• 好处 : 统一元数据 , 如 : 直接用 Hive 元数据

Flink 四种 Catalog :

• GenericInMemoryCatalog: 基于内存实现 Catalog，只在 session 的生命周期 (一个任务) 。默认自动创建 default_catalog 内存 Catalog (区分大小写)
• JdbcCatalog : 通过 JDBC 协议连接到关系数据库 (PostgreSQL ,MySQL )
• HiveCatalog : 作 Flink 元数据的持久化存储 ; 作 Hive 元数据的接口。(统一小写)
• 用户自定义 Catalog :通过 CatalogFactory.Context#getClassLoader 访问

JdbcCatalog

Flink 通过 JdbcCatalog 不能建表

• 只能在 MySQL 读写库表

cp flink-connector-jdbc-1.17-20230109.003314-120.jar  /opt/module/flink-1.17.0/lib/
cp mysql-connector-j-8.0.31.jar /opt/module/flink-1.17.0/lib/

jdbc 连接 : https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-connector-jdbc/1.17-SNAPSHOT/

创建 Catalog

• name : Catalog 名
• default-database : 连接到的默认数据库
• username : Postgres/MySQL 帐户的用户名
• password : 密码
• base-url : 数据库的 jdbc url
  • Postgres Catalog : jdbc:postgresql://<ip>:<端口>
  • MySQL Catalog : jdbc: mysql://<ip>:<端口>

CREATE CATALOG my_jdbc_catalog WITH(
    'type' = 'jdbc',
    'default-database' = 'test',
    'username' = 'root',
    'password' = '000000',
    'base-url' = 'jdbc:mysql://hadoop102:3306'
);

查看 Catalog :

-- 查看所有 Catalog
SHOW CATALOGS;

-- 查看当前的 CATALOG
SHOW CURRENT CATALOG;


-- 用指定 Catalog
USE CATALOG my_jdbc_catalog;

HiveCatalog

配置 :

# jar包
cp flink-sql-connector-hive-3.1.3_2.12-1.17.0.jar /opt/module/flink-1.17.0/lib/

cp mysql-connector-j-8.0.31.jar /opt/module/flink-1.17.0/lib/


# 更换 planner 依赖
mv /opt/module/flink-1.17.0/opt/flink-table-planner_2.12-1.17.0.jar /opt/module/flink-1.17.0/lib/flink-table-planner_2.12-1.17.0.jar

mv /opt/module/flink-1.17.0/lib/flink-table-planner-loader-1.17.0.jar /opt/module/flink-1.17.0/opt/flink-table-planner-loader-1.17.0.jar

启动 Hive metastore :

hive --service metastore &

创建 Catalog :

• type : Catalog 类型
• name : Catalog 名
• hive-conf-dir : hive-site.xml 目录 (选填)
• default-database : 默认数据库 (选填)
• hive-version : 指定 Hive 版本 (选填)
• hadoop-conf-dir : Hadoop conf 目录路径 (选填)

CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/opt/module/hive/conf'
);

读写 Hive 表 :

-- 查看 hive 数据库
SHOW DATABASES; 

-- 切到 hive 数据库
USE test;

-- 查看 hive 表
SHOW TABLES;

-- 读取 hive 表
SELECT * from ws;

-- 写入 hive 表
INSERT INTO ws VALUES(1, 1, 1);