详解 Flink Catalog 在 ChunJun 中的实践之路

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
实时计算 Flink 版,5000CU*H 3个月
云数据库 RDS MySQL,高可用系列 2核4GB
简介: Catalog 提供元数据,以及访问存储在数据库或其他外部系统中的数据所需的函数和信息,能够极大简化用户现有系统开始使用 Flink 所需的步骤,并增强用户体验,本文将为大家带来 Flink Catalog 的介绍以及 Flink Catalog 在 ChunJun 中的实践之路详解。

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

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

Flink Catalog 简介

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

Flink Catalog 作用

数据处理中最关键的一个方面是管理元数据:

· 可能是暂时性的元数据,如临时表,或针对表环境注册的 UDFs;

· 或者是永久性的元数据,比如 Hive 元存储中的元数据。

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

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

Flink Catalog 的结构

● Flink Catalog 原生结构

• GenericInMemoryCatalog:基于内存实现的 Catalog

• Jdbc Catalog:可以将 Flink 通过 JDBC 协议连接到关系数据库,目前 Flink 在1.12和1.13中有不同的实现,包括 MySql Catalog 和 Postgres Catalog

• Hive Catalog:作为原生 Flink 元数据的持久化存储,以及作为读写现有 Hive 元数据的接口

● Flink Iceberg Catalog

● Flink Hudi Catalog

HoodieCatalog、HoodieHiveCatalog

image.png

image.png

Flink Catalog 详解

GenericInMemoryCatalog

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 里面,无法持久化。

JDBC Catalog

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

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

JDBC Catalog 支持以下参数:

• name:必填,Catalog 的名称

• default-database:必填,默认要连接的数据库

• username:必填,Postgres/MySQL 账户的用户名

• password:必填,账户的密码

• base-url: 必填,(不应该包含数据库名)

对于 Postgres Catalog base-url 应为 "jdbc:postgresql://:" 的格式

对于 MySQL   Catalog base-url 应为 "jdbc:mysql://:" 的格式

Hive 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 session
USE CATALOG myhive;

image.png

Iceberg Catalog

● Hive Catalog 管理 Iceberg 表

(Flink) default_database.flink_table -> 
(Iceberg) default_database.flink_table
CREATE 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_table
CREATE 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'
);

● Hadoop Catalog 管理 Iceberg 表

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'
);

● 自定义 Catalog 管理 Iceberg 表

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',
     ...
);

• connector:iceberg

• catalog-name:用户指定的目录名称,这是必须的,因为连接器没有任何默认值

• catalog-type:内置目录的 hive 或 hadoop(默认为hive),或者对于使用 catalog-impl 的自定义目录实现,不做设置

• catalog-impl:自定义目录实现的全限定类名,如果 catalog-type 没有被设置,则必须被设置,更多细节请参见自定义目录

• catalog-database: 后台目录中的 iceberg 数据库名称,默认使用当前的 Flink 数据库名称

• catalog-table: 后台目录中的冰山表名,默认使用 Flink CREATE TABLE 句子中的表名

Hudi Catalog

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;

image.png

image.png

Flink Catalog 在 ChunJun 中的实践

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

直接引入开源 Catalog

ChunJun 目前的所有 Catalog 为以下四种:

image.png

● Hive Catalog 需要的依赖

image.png

● Iceberg Catalog 需要的依赖

image.png

● JDBC Catalog

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

● DT Catalog

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

DT Catalog -存储元数据表设计

● 创建 mysql 元数据表 database_info

-- 创建表的 sql
create 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`);

● 创建 mysql 元数据表 table_info

-- 创建表的 sql
create 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`);

● 创建 mysql 元数据表 properties_info

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,

使用 DT Catalog

● 创建 DT Catalog

CREATE CATALOG catalog1
WITH (
    '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'
  );

image.png

● 创建 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 EXISTS
If the database does not exist, nothing happens.
RESTRICT
Dropping a non-empty database triggers an exception. Enabled by default.
CASCADE
Dropping 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

● 创建 Table

1)Rename Table

ALTER TABLE [catalog_name.][db_name.]table_name RENAME TO new_table_name
Rename 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 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'
);
-- 删除表
drop table if exists mysql_catalog2.wujuan_database2.wujuan_table
-- 重命名表名
ALTER TABLE catalog1.default_database.table1 RENAME TO table2;
-- 设置表属性
ALTER TABLE catalog1.default_database.table1 
SET (
'tablename'='t2',
'url'='dbc:mysql://172.16.83.218:3306/wujuan?useSSL=false'
)

使用 DTCatalog 的具体场景和实现原理

● 全部是 DDL,只有 Catalog 的创建

· 可以执行,但是没有意义,ChunJun 不会存储 Catalog 信息,只有平台存储;

· 不支持语法校验。

CREATE CATALOG catalog1
WITH (
    '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'
  );

● 全部是 DDL,包含 Catalog、Database、Table 的创建

· 无论创建数据库、表,删除数据库、表,必须包含 create catalog 语句;

· 可以执行,可以创建数据库和表;

· 不支持语法校验。

-- 初始化 Catalog
CREATE CATALOG catalog1
WITH (
    '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;
    }
}

image.png

● DDL + DML,包含 create + insert 语句

1)初始化 Catalog

CREATE CATALOG catalog1
WITH (
    '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'
);

3.1)创建数据库

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'
);

4)执行任务

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

· 不可以执行,可以提交;

· 支持语法校验。

● DML,只有 Insert 语句

-- 初始化 Catalog
CREATE CATALOG catalog1
WITH (
  '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
· 如果 Catalog 的 数据库和表都已经创建好了,那么直接写 insert 就可以提交任务;

· 不可以执行,可以提交;

· 支持语法校验。


《数据治理行业实践白皮书》下载地址:https://fs80.cn/380a4b

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友,浏览袋鼠云官网:https://www.dtstack.com/?src=szalykfz

同时,欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术 qun」,交流最新开源技术信息,qun 号码:30537511,项目地址:https://github.com/DTStack

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
目录
相关文章
|
5月前
|
存储 监控 数据挖掘
京东物流基于Flink & StarRocks的湖仓建设实践
本文整理自京东物流高级数据开发工程师梁宝彬在Flink Forward Asia 2024的分享,聚焦实时湖仓的探索与建设、应用实践、问题思考及未来展望。内容涵盖京东物流通过Flink和Paimon等技术构建实时湖仓体系的过程,解决复杂业务场景下的数据分析挑战,如多维OLAP分析、大屏监控等。同时,文章详细介绍了基于StarRocks的湖仓一体方案,优化存储成本并提升查询效率,以及存算分离的应用实践。最后,对未来数据服务的发展方向进行了展望,计划推广长周期数据存储服务和原生数据湖建设,进一步提升数据分析能力。
420 1
京东物流基于Flink & StarRocks的湖仓建设实践
|
3月前
|
资源调度 Kubernetes 流计算
Flink在B站的大规模云原生实践
本文基于哔哩哔哩资深开发工程师丁国涛在Flink Forward Asia 2024云原生专场的分享,围绕Flink On K8S的实践展开。内容涵盖五个部分:背景介绍、功能及稳定性优化、性能优化、运维优化和未来展望。文章详细分析了从YARN迁移到K8S的优势与挑战,包括资源池统一、环境一致性改进及隔离性提升,并针对镜像优化、Pod异常处理、启动速度优化等问题提出解决方案。此外,还探讨了多机房容灾、负载均衡及潮汐混部等未来发展方向,为Flink云原生化提供了全面的技术参考。
194 9
Flink在B站的大规模云原生实践
|
4月前
|
SQL 存储 NoSQL
Flink x Paimon 在抖音集团生活服务的落地实践
本文整理自抖音集团数据工程师陆魏与流式计算工程冯向宇在Flink Forward Asia 2024的分享,聚焦抖音生活服务业务中的实时数仓技术演变及Paimon湖仓实践。文章分为三部分:背景及现状、Paimon湖仓实践与技术优化。通过引入Paimon,解决了传统实时数仓开发效率低、资源浪费、稳定性差等问题,显著提升了开发运维效率、节省资源并增强了任务稳定性。同时,文中详细探讨了Paimon在维表实践、宽表建设、标签变更检测等场景的应用,并介绍了其核心技术优化与未来规划。
412 10
Flink x Paimon 在抖音集团生活服务的落地实践
|
4月前
|
资源调度 Kubernetes 调度
网易游戏 Flink 云原生实践
本文分享了网易游戏在Flink实时计算领域的资源管理与架构演进经验,从Yarn到K8s云原生,再到混合云的实践历程。文章详细解析了各阶段的技术挑战与解决方案,包括资源隔离、弹性伸缩、自动扩缩容及服务混部等关键能力的实现。通过混合云架构,网易游戏显著提升了资源利用率,降低了30%机器成本,小作业计算成本下降40%,并为未来性能优化、流批一体及智能运维奠定了基础。
226 9
网易游戏 Flink 云原生实践
|
6月前
|
存储 运维 监控
阿里妈妈基于 Flink+Paimon 的 Lakehouse 应用实践
本文总结了阿里妈妈数据技术专家陈亮在Flink Forward Asia 2024大会上的分享,围绕广告业务背景、架构设计及湖仓方案演进展开。内容涵盖广告生态运作、实时数仓挑战与优化,以及基于Paimon的湖仓方案优势。通过分层设计与技术优化,实现业务交付周期缩短30%以上,资源开销降低40%,并大幅提升系统稳定性和运营效率。文章还介绍了阿里云实时计算Flink版的免费试用活动,助力企业探索实时计算与湖仓一体化解决方案。
769 3
阿里妈妈基于 Flink+Paimon 的 Lakehouse 应用实践
|
6月前
|
存储 SQL Java
Flink CDC + Hologres高性能数据同步优化实践
本文整理自阿里云高级技术专家胡一博老师在Flink Forward Asia 2024数据集成(二)专场的分享,主要内容包括:1. Hologres介绍:实时数据仓库,支持毫秒级写入和高QPS查询;2. 写入优化:通过改进缓冲队列、连接池和COPY模式提高吞吐量和降低延迟;3. 消费优化:优化离线场景和分区表的消费逻辑,提升性能和资源利用率;4. 未来展望:进一步简化用户操作,支持更多DDL操作及全增量消费。Hologres 3.0全新升级为一体化实时湖仓平台,提供多项新功能并降低使用成本。
485 1
Flink CDC + Hologres高性能数据同步优化实践
|
6月前
|
SQL 存储 调度
基于 Flink 进行增量批计算的探索与实践
基于 Flink 进行增量批计算的探索与实践
125 1
基于 Flink 进行增量批计算的探索与实践
|
6月前
|
存储 运维 BI
万字长文带你深入广告场景Paimon+Flink全链路探索与实践
本文将结合实时、离线数据研发痛点和当下Paimon的特性,以实例呈现低门槛、低成本、分钟级延迟的流批一体化方案,点击文章阅读详细内容~
|
6月前
|
SQL 弹性计算 DataWorks
Flink CDC 在阿里云 DataWorks 数据集成入湖场景的应用实践
Flink CDC 在阿里云 DataWorks 数据集成入湖场景的应用实践
237 6
|
6月前
|
SQL 存储 HIVE
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
360 2