FlinkSQL 中 Catalog 的使用場景及案例詳解
作者：甜獲師兄
Catalog 在 Flink SQL 中是一個元數據管理組件，用于存儲和管理數據庫、表、視 圖、函數等元數據對象的抽象接口。
Catalog 在 Flink SQL 中是一個元數據管理組件，用于存儲和管理數據庫、表、視 圖、函數等元數據對象的抽象接口。它類似于傳統數據庫系統中的元數據倉庫，為 Flink SQL 提供了元數據管理能力。
Catalog 使 Flink 能夠：
- 以統一的方式訪問不同的外部系統
- 減少代碼中的硬編 碼配置
- 實現表元數據的持久化
- 支持跨會話的元數據共享。
1. Catalog的作用
(1) 管理元數據對象
Catalog 可以管理以下元數據對象： - 數據庫(Database) - 表(Table) - 視圖(View) - 函數(Function) - 分區(Partition)等
(2) 支持多樣化的元數據存儲
Flink 支持多種Catalog 實現，可以連接各種外部元數據系統： - 內存Catalog(默認) - Hive Catalog - JDBC Catalog - 自定義 Catalog
(3) 提供統一的數據訪問接口
無論底層元數據存儲在哪里，都可以通過統一的接口訪問和操作
(4) 簡化元數據管理
開發者可以通過Catalog 注冊永久表，而不是在代碼中重復定義表結構
2. Flink內置的catalog類型
(1) GenericInMemoryCatalog
默認的內存 Catalog，元數據只在單個 Flink 會話中有效，會話結束數據就會丟失。
// 創建內存Catalog
Catalog inmemory = new GenericInMemoryCatalog("in_memory_catalog");
tableEnv.registerCatalog("in_memory_catalog", inmemory); -- SQL 中創建和使用內存Catalog
CREATE CATALOG my_memory_catalog WITH (
'type' = 'generic_in_memory'
);
USE CATALOG my_memory_catalog;(2) HiveCatalog
使用Hive Metastore 存儲元數據，支持持久化，適合生產環境。
// 創建Hive Catalog
String name = "my_hive_catalog";
String defaultDatabase = "default";
String hiveConfDir = "/path/to/hive/conf";
String version = "2.3.6";
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir,
version);
tableEnv.registerCatalog("my_hive_catalog", hive); -- SQL 中創建和使用Hive Catalog
CREATE CATALOG my_hive_catalog WITH (
'type' = 'hive',
'default-database' = 'default',
'hive-conf-dir' = '/path/to/hive/conf',
'hive-version' = '2.3.6'
);
USE CATALOG my_hive_catalog;
3. JdbcCatalog
使用關系型數據庫存儲元數據。
// 創建JDBC Catalog
String name = "my_jdbc_catalog";
String defaultDatabase = "default";
String username = "username";
String password = "password";
String baseUrl = "jdbc:mysql://localhost:3306";
JdbcCatalog jdbc = new JdbcCatalog(name, defaultDatabase, username, pas
sword, baseUrl);
tableEnv.registerCatalog("my_jdbc_catalog", jdbc); -- SQL中創建和使用JDBC Catalog(Flink 1.15+)
CREATE CATALOG my_jdbc_catalog WITH (
'type' = 'jdbc',
'default-database' = 'default',
'username' = 'username',
'password' = 'password',
'base-url' = 'jdbc:mysql://localhost:3306'
);
USE CATALOG my_jdbc_catalog;(3) 使用Catalog的SQL操作
1. 創建和切換Catalog -- 創建Catalog
CREATE CATALOG my_catalog WITH (
'type' = 'hive',
'hive-conf-dir' = '/path/to/hive/conf'
);
-- 查看所有Catalog
SHOW CATALOGS;
-- 切換當前Catalog
USE CATALOG my_catalog;
2. 創建和使用數據庫 -- 創建數據庫
CREATE DATABASE my_database;
-- 查看所有數據庫
SHOW DATABASES;
-- 切換當前數據庫
USE my_database;
3. 管理表和視圖 -- 創建表
CREATE TABLE my_table (
id INT,
name STRING,
create_time TIMESTAMP(3)
) WITH (
'connector' = 'kafka',
'topic' = 'my_topic',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
); -- 查看所有表
SHOW TABLES; -- 查看表結構
DESCRIBE my_table;
4. 管理函數 -- 創建自定義函數
CREATE FUNCTION my_function AS 'com.example.MyFunction'; -- 查看所有函數
SHOW FUNCTIONS;(4) Catalog 的實際應用示
// 跨會話持久化元數據
// 會話1：注冊Hive Catalog和表
StreamExecutionEnvironment env1 = StreamExecutionEnvironment.getExecuti
onEnvironment();
StreamTableEnvironment tEnv1 = StreamTableEnvironment.create(env1);
tEnv1.executeSql("CREATE CATALOG hive_catalog WITH ('type' = 'hive')");
tEnv1.executeSql("USE CATALOG hive_catalog");
tEnv1.executeSql("CREATE DATABASE IF NOT EXISTS db1");
tEnv1.executeSql("USE db1");
tEnv1.executeSql(
"CREATE TABLE IF NOT EXISTS orders (" +
" order_id STRING, " +
" price DECIMAL(10, 2)" +
") WITH (" +
" 'connector' = 'kafka', " +
" 'topic' = 'orders'" +
")");
// 會話2：直接使用之前注冊的表
StreamExecutionEnvironment env2 = StreamExecutionEnvironment.getExecuti
onEnvironment();
StreamTableEnvironment tEnv2 = StreamTableEnvironment.create(env2);
tEnv2.executeSql("USE CATALOG hive_catalog");
tEnv2.executeSql("USE db1");
// 不需要重新定義表結構，可以直接查詢
tEnv2.executeSql("SELECT * FROM orders");
使用不同類型的Catalog實現多源數據集成
// 注冊多個Catalog訪問不同系統
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutio
nEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
// 注冊Hive Catalog
tEnv.executeSql("CREATE CATALOG hive_catalog WITH ('type' = 'hive')");
// 注冊JDBC Catalog
tEnv.executeSql(
"CREATE CATALOG jdbc_catalog WITH (" +
" 'type' = 'jdbc', " +
" 'default-database' = 'default', " +
" 'username' = 'user', " +
" 'password' = 'password', " +
" 'base-url' = 'jdbc:mysql://localhost:3306'" +
")");
// 從不同Catalog中的表關聯查詢
tEnv.executeSql(
"SELECT o.order_id, o.price, c.name " +
"FROM hive_catalog.db1.orders o " +
"JOIN jdbc_catalog.default.customers c " +
"ON o.customer_id = c.id"
); 責任編輯：趙寧寧
來源：
大數據技能圈
