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

2023-04-07 15:09 作者:袋鼠云 0人读过 | 我要投稿

我们知道 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

Flink Catalog 详解

GenericInMemoryCatalog

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

JDBC 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

Iceberg Catalog

● Hive Catalog 管理 Iceberg 表

● Hadoop Catalog 管理 Iceberg 表

● 自定义 Catalog 管理 Iceberg 表

• 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

Flink Catalog 在 ChunJun 中的实践

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

直接引入开源 Catalog

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

● Hive Catalog 需要的依赖

● Iceberg Catalog 需要的依赖

● JDBC Catalog

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

● DT Catalog

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

DT Catalog -存储元数据表设计

● 创建 mysql 元数据表 database_info

● 创建 mysql 元数据表 table_info

● 创建 mysql 元数据表 properties_info

● properties_info 里面存了什么?

使用 DT Catalog

● 创建 DT Catalog

● 创建 Database

● 创建 Table

1）Rename Table

2）Set or Alter Table Properties

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

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

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

· 不支持语法校验。

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

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

· 可以执行，可以创建数据库和表；

· 不支持语法校验。

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

1）初始化 Catalog

2.1）创建数据库

2.2）创建源表

3.1）创建数据库

3.2）创建结果表

4）执行任务

· 不可以执行，可以提交；

· 支持语法校验。

● DML，只有 Insert 语句

· 如果 Catalog 的数据库和表都已经创建好了，那么直接写 insert 就可以提交任务；