MySQL Catalog 提供统一的元数据管理和数据访问能力，支持 Flink 作业直接读写 MySQL 数据库中的表。创建 Catalog 后，您可以在 Flink SQL 中使用三段式表名（catalog.database.table）访问 MySQL 数据表，无需在每个作业中重复配置连接参数。

创建 MySQL Catalog 时，需要选择一个 Flink Session 集群。该集群用于访问目标 MySQL 数据库、执行连通性测试，并同步库表元数据。元数据同步完成后，Flink SQL 作业可以通过 Catalog 查询或读写 MySQL 表。
MySQL Catalog 采用手动同步机制。MySQL 侧新增、删除或修改库表结构后，需要再次单击 同步 Catalog，将最新的库表元数据同步到 Flink 侧。

• MySQL Catalog 仅支持在 Flink 1.17-volcano 及以上引擎版本中使用。
• 当前版本仅支持 MySQL 5.7 及以上版本的数据库。
• Flink Session 集群所在网络必须能够访问目标 MySQL 数据库实例。

MySQL 侧 #

• 已创建目标 MySQL 数据库实例，并获取数据库连接地址（Hostname）和端口号（Port，默认为 3306）。
• 已创建具有目标数据库读写权限的数据库用户，并获取用户名和密码。

Flink 侧 #

• 已创建用于连接目标 MySQL 数据库和同步库表元数据的 Flink Session 集群。
• Session 集群的引擎版本为 Flink 1.17-volcano 或更高版本。
• Session 集群与 MySQL 实例之间网络互通，且安全组已放行 MySQL 实例的连接端口。

1. 在数据目录页面单击 创建 Catalog。选择 MySQL 类型的 Catalog，然后单击 下一步。
   

2. 填写 Catalog 配置信息，各参数说明如下：

   参数名称	配置说明
   Catalog 类型	显示上一步选择的数据目录类型，此处为 MySQL。
   Catalog 名称	用于标识该数据目录的名称，例如 mysql_catalog。
   Hostname	MySQL 数据库的连接地址，即主机名。请填写实例的实际连接地址。
   Port	MySQL 数据库的连接端口。页面提示值为 3306，请以实例的实际端口为准。
   Default database	Catalog 默认使用的数据库名称，例如 test_db。
   Username	用于连接 MySQL 数据库的用户名，请填写实际数据库账号。
   Password	数据库用户对应的密码。根据现有技术方案说明，该配置将加密存储。
   Session 集群	用于访问 MySQL、执行连通性测试和同步库表元数据的 Flink Session 集群。所选集群必须能访问目标数据库，并满足引擎版本要求。没有可用集群时，可通过页面上的 创建新 Session 集群 入口创建；创建完成后返回当前页面并刷新列表，再从下拉列表中选择。

3. 填写完成后，单击 连通性测试，验证数据库连接是否正常。

4. 连通性测试通过后，单击 下一步，完成 Catalog 创建。

同步库表元数据 #

1. Catalog 创建成功后，在数据目录页面找到目标 Catalog。单击 同步 Catalog，同步 MySQL 库表元数据。
   
2. 同步完成后，在当前页面查看已同步的库表元数据信息。
   

查看、编辑或删除 Catalog #

在 Catalog 列表中找到目标 Catalog，然后根据需要单击 查看、编辑 或 删除。

在 Flink SQL 任务中，可以使用 catalog.database.table 三段式表名访问 MySQL 数据表。以下示例中的 Catalog、数据库和表名称需替换为实际名称。

读取 MySQL 表 #

SELECT *
FROM `mysql_catalog`.`test_db`.`test_tb`;

向 MySQL 表写入数据 #

-- 模拟数据写入
CREATE TABLE orders (
  t1 INT,
  t2 STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1'
);

-- 写入 MySQL 表
INSERT INTO `mysql_catalog`.`test_db`.`orders_sink`
SELECT t1, t2 FROM orders;

• 保存 Flink SQL 任务，配置任务资源并发布上线。
• 启动任务，确认任务运行正常。
• 如执行写入操作，请在 MySQL 数据库中检查目标表，确认数据已成功写入。

连通性测试失败 #

现象：​创建 Catalog 时，连通性测试不通过。
排查步骤：

• 检查 Hostname（数据库连接地址）和 Port（端口号）是否填写正确。
• 检查 Username（用户名）和 Password（密码）是否正确。
• 确认 Session 集群所在网络与 MySQL 实例之间网络互通，且安全组已放行对应端口。
• 确认 Session 集群的引擎版本为 Flink 1.17-volcano 或更高版本。

数据目录同步失败 #

现象：​连通性测试成功，但数据目录同步失败，或 MySQL 表结构变更未同步到 Catalog。
排查步骤：

• 检查当前数据库用户是否具有目标数据库的读权限。
• 确认用于同步的 Session 集群处于运行状态。
• 确认 Session 集群与 MySQL 实例之间网络互通。
• 如果 MySQL 表结构已发生变更，请返回数据目录页面，再次单击 同步 Catalog。同步后重新查看库表元数据信息。