我们知道 Flink 有Table（表）、View（视图）、Function（函数/算子）、Database（数据库）的概念，相对于这些耳熟能详的概念，Flink 里还有一个 Catalog（目录） 的概念。

本文将为大家带来 Flink Catalog 的介绍以及 Flink Catalog 在 ChunJun 中的实践之路。

Catalog 提供元数据，如数据库、表、分区、视图，以及访问存储在数据库或其他外部系统中的数据所需的函数和信息。

Catalog 提供了一个统一的 API 来管理元数据，并使其可以从表 API 和 SQL 查询语句中来访问。

Catalog 使用户能够引用他们数据系统中的现有元数据，并自动将它们映射到 Flink 的相应元数据。例如，Flink 可以将 JDBC 表自动映射到 Flink 表，用户不必在 Flink 中手动重写 DDL。Catalog 大大简化了用户现有系统开始使用 Flink 所需的步骤，并增强了用户体验。

final CatalogManager catalogManager =        CatalogManager.newBuilder()                .classLoader(userClassLoader)                .config(tableConfig)                .defaultCatalog(                        settings.getBuiltInCatalogName(),                        new GenericInMemoryCatalog(                                settings.getBuiltInCatalogName(),                                settings.getBuiltInDatabaseName()))                .build();
defaultCatalog =                    new GenericInMemoryCatalog(                            defaultCatalogName, settings.getBuiltInDatabaseName());
CatalogManager catalogManager =                builder.defaultCatalog(defaultCatalogName, defaultCatalog).build();

GenericInMemoryCatalog 所有的数据都保存在 HashMap 里面，无法持久化。

CREATE CATALOG my_catalog WITH(    'type' = 'jdbc',    'default-database' = '...',    'username' = '...',    'password' = '...',    'base-url' = '...');
USE CATALOG my_catalog;

如果创建并使用 Postgres Catalog 或 MySQL Catalog，请配置 JDBC 连接器和相应的驱动。

JDBC Catalog 支持以下参数:

CREATE CATALOG myhive WITH (    'type' = 'hive',    'default-database' = 'mydatabase',    'hive-conf-dir' = '/opt/hive-conf');-- set the HiveCatalog as the current catalog of the sessionUSE CATALOG myhive;

● Hive Catalog 管理 Iceberg 表

(Flink) default_database.flink_table -> (Iceberg) default_database.flink_tableCREATE TABLE flink_table (    id   BIGINT,    data STRING) WITH (    'connector'='iceberg',    'catalog-name'='hive_prod',    'uri'='thrift://localhost:9083',    'warehouse'='hdfs://nn:8020/path/to/warehouse');(Flink)default_database.flink_table -> (Iceberg) hive_db.hive_iceberg_tableCREATE TABLE flink_table (    id   BIGINT,    data STRING) WITH (    'connector'='iceberg',    'catalog-name'='hive_prod',    'catalog-database'='hive_db',    'catalog-table'='hive_iceberg_table',    'uri'='thrift://localhost:9083',    'warehouse'='hdfs://nn:8020/path/to/warehouse');

CREATE TABLE flink_table (    id   BIGINT,    data STRING) WITH (    'connector'='iceberg',    'catalog-name'='hadoop_prod',    'catalog-type'='hadoop',    'warehouse'='hdfs://nn:8020/path/to/warehouse');

CREATE TABLE flink_table (    id   BIGINT,    data STRING) WITH (    'connector'='iceberg',    'catalog-name'='custom_prod',    'catalog-impl'='com.my.custom.CatalogImpl',     -- More table properties for the customized catalog    'my-additional-catalog-config'='my-value',     ...);

create catalog hudi with( 'type' = 'hudi', 'mode' = 'hms',  'hive.conf.dir'='/etc/hive/conf');
--- 创建数据库供hudi使用create database hudi.hudidb;
--- order表CREATE TABLE hudi.hudidb.orders_hudi(  uuid INT,  ts INT,  num INT,  PRIMARY KEY(uuid) NOT ENFORCED) WITH (  'connector' = 'hudi',  'table.type' = 'MERGE_ON_READ');
select * from hudi.hudidb.orders_hudi;

下面将为大家介绍本文的重头戏，Flink Catalog 在 ChunJun 中的实践之路。

JDBC 因为 Flink 1.12 和 1.13 API 有变化，因此需要涉及源码的改动，改动一些 API 后，从源码引入。

结合内部业务，自定义的一种 Catalog ，下文将会进行详细介绍。

● 创建 mysql 元数据表 database_info

-- 创建表的 sqlcreate table database_info(    `id`            bigint PRIMARY KEY NOT NULL AUTO_INCREMENT COMMENT '项目ID',-- database id    `catalog_name`  varchar(255) COMMENT 'catalog 名字',    `database_name` varchar(255) COMMENT 'database 名字',    `catalog_type`  varchar(30) COMMENT 'catalog 类型, eg: mysql,oracle...',    `project_id`    int(11)            NOT NULL COMMENT '项目ID',    `tenant_id`     int(11)            NOT NULL COMMENT '租户ID') ENGINE = InnoDB  DEFAULT CHARSET = utf8;
-- 创建索引CREATE INDEX idx_catalog_name_database_name_project_id_tenant_id ON database_info (`catalog_name`, `database_name`, `project_id`, `tenant_id`);

-- 创建表的 sqlcreate table table_info(    `id`            bigint PRIMARY KEY NOT NULL AUTO_INCREMENT,    `database_id`    bigint COMMENT 'database_info 表的 id',    `table_name`  varchar(255) COMMENT '表名',    `project_id`    int(11)            NOT NULL COMMENT '项目ID',    `tenant_id`     int(11)            NOT NULL COMMENT '租户ID') ENGINE = InnoDB  DEFAULT CHARSET = utf8;
-- 创建索引CREATE INDEX idx_catalog_id_project_id_tenant_id ON table_info (`database_id`, `project_id`, `tenant_id`);CREATE INDEX idx_database_id_table_name_project_id_tenant_id ON table_info (`database_id`, `table_name`, `project_id`, `tenant_id`);

create table properties_info(    `id`       bigint PRIMARY KEY NOT NULL AUTO_INCREMENT ,    `table_id` bigint(20) COMMENT 'table_info 表的 id',    `key`      varchar(255) COMMENT '表的属性 key',    `value`    varchar(255) COMMENT '表的属性 value') ENGINE = InnoDB  DEFAULT CHARSET = utf8;
CREATE INDEX idx_table_id ON properties_info (table_id);

● properties_info 里面存了什么?

schema.0.name=id,  schema.0.data-type=INT NOT NULL,  schema.1.name=name,  schema.1.data-type=VARCHAR(2147483647)  schema.2.name=age,  schema.2.data-type=BIGINT,  schema.primary-key.name=PK_3386,  schema.primary-key.columns=id,
  connector=jdbc,  url=jdbc:mysql: //172.16.83.218:3306/wujuan?useSSL=false,  username=drpeco,  password=DT@Stack#123,
  comment=,  scan.auto-commit=true,  lookup.cache.max-rows=20000,  scan.fetch-size=10,  lookup.cache.ttl=700000  table-name=t2,

CREATE CATALOG catalog1WITH (    'type' = 'dt',    'default-database' = 'default_database',    'driver' = 'com.mysql.cj.jdbc.Driver',    'url' = 'jdbc:mysql://xxx:3306/catalog_default',    'username' = 'drpeco',    'password' = 'DT@Stack#123',    'project-id' = '1',    'tenant-id' = '1'  );

● 创建 Database

DROP DATABASE [IF EXISTS] [catalog_name.]db_name [ (RESTRICT | CASCADE) ]Drop a database with the given database name. If the database to drop does not exist, an exception is thrown.IF EXISTSIf the database does not exist, nothing happens.RESTRICTDropping a non-empty database triggers an exception. Enabled by default.CASCADEDropping a non-empty database also drops all associated tables and functions.

create database if not exists catalog1.database1
drop database if exists catalog1.database1 -- 删除非空数据库，连通数据库中的所有表也一起删除drop database if exists catalog1.database1 CASCADE

1）Rename Table

ALTER TABLE [catalog_name.][db_name.]table_name RENAME TO new_table_nameRename the given table name to another new table name

2）Set or Alter Table Properties

ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...) Set one or more properties in the specified table. If a particular property is already set in the table, override the old value with the new one.

CREATE CATALOG catalog1WITH (    'type' = 'DT',    'default-database' = 'default_database',    'driver' = 'com.mysql.cj.jdbc.Driver',    'url' = 'jdbc:mysql://172.16.100.186:3306/catalog_default?autoReconnect=true&failOverReadOnly=false',    'username' = 'drpeco',    'password' = 'DT@Stack#123',    'project-id' = '1',    'tenant-id' = '1'  );

· 不支持语法校验。

-- 初始化 CatalogCREATE CATALOG catalog1WITH (    'type' = 'dt',    'default-database' = 'default_database',    'driver' = 'com.mysql.cj.jdbc.Driver',    'url' = 'jdbc:mysql://172.16.100.186:3306/catalog_default',    'username' = 'drpeco',    'password' = 'DT@Stack#123',    'project-id' = '1',    'tenant-id' = '1'  );-- 创建数据库create database if not exists database1-- 创建表CREATE TABLE if not exists catalog1.default_database.table1(    id      int,    name    string,    age     bigint,    primary key ( id) not enforced) with (    'connector' = 'jdbc',    'url' = 'jdbc:mysql://172.16.83.218:3306/wujuan?useSSL=false',    'table-name' = 't2',    'username' = 'drpeco',    'password' = 'DT@Stack#123');

· 不支持语法校验。

// 抛出异常的逻辑StatementSet statementSet = SqlParser.parseSql(job, jarUrlList, tEnv);TableResult execute = statementSet.execute();       -->tableEnvironment.executeInternal(operations);        -->Pipeline pipeline = execEnv.createPipeline(transformations, tableConfig, jobName);     -->StreamGraph streamGraph = ExecutorUtils.generateStreamGraph(getExecutionEnvironment(), transformations); -->
// 抛出异常的方法public static StreamGraph generateStreamGraph(StreamExecutionEnvironment execEnv, List<Transformation<?>> transformations){        if (transformations.size() <= 0) {                throw new IllegalStateException(                        "No operators defined in streaming topology. Cannot generate StreamGraph.");        }        ...        return generator.generate();}

// 如果没有 insert 语句的时候，无法生成 JobGraph，但是 DDL 是执行成功的。// 因此捕获 FlinkX 抛出的特殊异常，此语句的异常 Message 是 FlinkX 里面处理的。try {    PackagedProgramUtils.createJobGraph(program, flinkConfig, 1, false);} catch (ProgramInvocationException e) {    // 仅执行 DDL FlinkX 抛出的异常    if (!e.getMessage().contains("OnlyExecuteDDL")) {        throw e;    }}

CREATE CATALOG catalog1WITH (    'type' = 'dt',    'default-database' = 'default_database',    'driver' = 'com.mysql.cj.jdbc.Driver',    'url' = 'jdbc:mysql://172.16.100.186:3306/catalog_default',    'username' = 'drpeco',    'password' = 'DT@Stack#123',    'project-id' = '1',    'tenant-id' = '1'  );

2.1）创建数据库

create database if not exists database1

2.2）创建源表

CREATE TABLE if not exists catalog1.default_database.table1(    id      int,    name    string,    age     bigint,    primary key ( id) not enforced) with (    'connector' = 'jdbc',    'url' = 'jdbc:mysql://172.16.83.218:3306/wujuan?useSSL=false',    'table-name' = 't2',    'username' = 'drpeco',    'password' = 'DT@Stack#123');

create database if not exists catalog1.database2;

3.2）创建结果表

CREATE TABLE if not exists catalog1.database2.table2(    id      int,    name    string,    age     bigint,    primary key ( id) not enforced) with ( 'connector' = 'print');

insert into catalog1.database2.table2 select * from catalog1.database1.table1

-- 初始化 CatalogCREATE CATALOG catalog1WITH (  'type' = 'dt',  'default-database' = 'default_database',  'driver' = 'com.mysql.cj.jdbc.Driver',  'url' = 'jdbc:mysql://172.16.100.186:3306/catalog_default',  'username' = 'drpeco',  'password' = 'DT@Stack#123',  'project-id' = '1',  'tenant-id' = '1');
-- 执行任务insert into catalog1.database2.table2 select * from catalog1.database1.table1