Flink CDC 中分库分表采集，这里都是说同一个mysql实例去搞，那么如果是多个myql实例？

在Flink CDC中实现多个MySQL实例的分库分表采集，可以通过以下方法来实现。以下是详细的步骤和配置说明：

1. 使用正则表达式匹配多个MySQL实例

Flink CDC支持通过正则表达式匹配多个数据库实例中的分库分表。您可以在CREATE TABLE语句中为每个MySQL实例分别定义数据源，并使用正则表达式匹配对应的库名和表名。

示例代码

假设您有两个MySQL实例（instance1 和 instance2），每个实例中都有分库分表的数据需要采集。可以按照以下方式配置：

-- 定义第一个MySQL实例的数据源
CREATE TABLE mysql_instance1_orders (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA FROM 'table_name' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'instance1-host', -- 第一个MySQL实例的主机名
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'db_.*', -- 正则匹配多个分库
    'table-name' = 'orders_.*' -- 正则匹配多张分表
);

-- 定义第二个MySQL实例的数据源
CREATE TABLE mysql_instance2_orders (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA FROM 'table_name' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'instance2-host', -- 第二个MySQL实例的主机名
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'db_.*', -- 正则匹配多个分库
    'table-name' = 'orders_.*' -- 正则匹配多张分表
);

注意： 每个MySQL实例需要单独定义数据源，并且hostname参数需要指向不同的MySQL实例地址。

2. 合并多个MySQL实例的数据到下游表

为了将多个MySQL实例的数据合并到下游的一张表中，可以使用INSERT INTO语句将不同实例的数据写入同一个目标表。

示例代码

假设目标表是Hologres中的holo_orders表，可以按照以下方式合并数据：

-- 将第一个MySQL实例的数据写入Hologres
INSERT INTO holo_orders
SELECT * FROM mysql_instance1_orders;

-- 将第二个MySQL实例的数据写入Hologres
INSERT INTO holo_orders
SELECT * FROM mysql_instance2_orders;

重要提示：
- 如果目标表需要区分数据来源，可以利用元数据列（如db_name和table_name）作为额外字段写入目标表。 - 确保目标表的主键设计能够保证数据唯一性，例如联合主键可以包含db_name、table_name和原主键字段。

3. 使用STATEMENT SET优化多个MySQL实例的同步

如果需要将多个MySQL实例的同步任务作为一个作业提交，可以使用STATEMENT SET语法来优化Source节点的复用，减少对数据库的连接数和读取压力。

示例代码

以下是一个使用STATEMENT SET的示例，将两个MySQL实例的数据同步到Hologres：

BEGIN STATEMENT SET;

-- 同步第一个MySQL实例的数据
INSERT INTO holo_orders
SELECT * FROM mysql_instance1_orders;

-- 同步第二个MySQL实例的数据
INSERT INTO holo_orders
SELECT * FROM mysql_instance2_orders;

END;

注意：
- 对于Source复用优化，需要确保这些Source表的WITH参数完全一致，才能成功合并复用。 - 如果不同MySQL实例的表结构不一致，可能需要在SQL中进行字段映射或转换。

4. 配置Binlog监听以支持增量同步

为了确保多个MySQL实例的增量数据能够被正确采集，需要在每个MySQL实例中开启Binlog，并确保Binlog过滤器未忽略相关库或表。

检查Binlog配置

在每个MySQL实例中执行以下命令，确认Binlog是否已正确配置：

SHOW MASTER STATUS;

输出示例：

+------------------+----------+--------------+------------------+-------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+-------------------+
| mysql-bin.000006 | 4594     |              |                  | xxx:1-15          |
+------------------+----------+--------------+------------------+-------------------+

重要提示：
- 如果Binlog_Ignore_DB或Binlog_Do_DB字段中忽略了某些库，则需要调整MySQL的Binlog配置。 - 确保每个MySQL实例的server-id范围不冲突，避免Binlog读取时出现重复或遗漏。

5. 资源配置与并发控制

对于多个MySQL实例的同步任务，建议根据数据量和业务需求合理配置并发数，以提高数据加载效率。

资源配置

• 在Flink实时计算控制台中，可以通过基础模式或专家模式设置作业的并发数：
  • 基础模式：设置整个作业的全局并发数。
  • 专家模式：按需为某个VERTEX设置并发数。

并发控制

• MySQL CDC支持多并发读取全量数据，但在增量阶段会自动缩容以节约资源。
• 建议结合Autopilot自动调优功能，动态调整并发数以适应数据流量变化。

总结

通过上述方法，您可以实现多个MySQL实例的分库分表采集，并将数据合并到下游的目标表中。关键点包括： 1. 为每个MySQL实例单独定义数据源，并使用正则表达式匹配分库分表。 2. 使用INSERT INTO语句将多个实例的数据写入同一张目标表。 3. 利用STATEMENT SET优化Source节点复用，减少数据库连接压力。 4. 确保每个MySQL实例的Binlog配置正确，并合理设置并发数以提高性能。

重要提醒：
- 确保主键唯一性：在合并多个MySQL实例的数据时，建议使用联合主键（如db_name、table_name和原主键）以避免数据冲突。 - 监控数据库压力：多个CDC作业可能会对MySQL实例造成较大压力，建议根据实际情况调整并发数或分批处理数据。