// 創(chuàng)建內(nèi)存Catalog 
Catalog inmemory = new GenericInMemoryCatalog("in_memory_catalog"); 
tableEnv.registerCatalog("in_memory_catalog", inmemory); -- SQL 中創(chuàng)建和使用內(nèi)存Catalog 
CREATE CATALOG my_memory_catalog WITH ( 
'type' = 'generic_in_memory' 
); 
USE CATALOG my_memory_catalog;

// 創(chuàng)建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 中創(chuàng)建和使用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 
使用關系型數(shù)據(jù)庫存儲元數(shù)據(jù)。 
// 創(chuàng)建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中創(chuàng)建和使用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;

1. 創(chuàng)建和切換Catalog -- 創(chuàng)建Catalog 
CREATE CATALOG my_catalog WITH ( 
    'type' = 'hive', 
    'hive-conf-dir' = '/path/to/hive/conf' 
); 
 -- 查看所有Catalog 
SHOW CATALOGS; 
 -- 切換當前Catalog 
USE CATALOG my_catalog; 
2. 創(chuàng)建和使用數(shù)據(jù)庫 -- 創(chuàng)建數(shù)據(jù)庫 
CREATE DATABASE my_database; 
 -- 查看所有數(shù)據(jù)庫 
SHOW DATABASES; 
 -- 切換當前數(shù)據(jù)庫 
USE my_database; 
3. 管理表和視圖 -- 創(chuàng)建表 
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. 管理函數(shù) -- 創(chuàng)建自定義函數(shù) 
CREATE FUNCTION my_function AS 'com.example.MyFunction'; -- 查看所有函數(shù) 
SHOW FUNCTIONS;

// 跨會話持久化元數(shù)據(jù) 
// 會話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實現(xiàn)多源數(shù)據(jù)集成 
// 注冊多個Catalog訪問不同系統(tǒng) 
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中的表關聯(lián)查詢 
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" 
);