如果Flink CDC 不用 CTAS 怎么把分库分表的 库名和表名同步过来？

在Flink CDC（Change Data Capture）中，如果不使用CTAS（Create Table As Select）语句来创建表，而是希望从分库分表的源数据库中同步库名和表名信息，可以通过以下步骤实现：

1. 定义源表: 首先，你需要为每个分库分表的源数据库定义一个源表。这个源表应该包含库名和表名的信息，通常可以使用database和table字段来表示。

2. 数据同步: 使用Flink CDC Connector来捕获源数据库的变更数据。你可以为每个源表配置一个CDC Connector，以便捕获其变更数据。

3. 合并数据: 你可能需要将所有源表的数据合并到一个目标表中。这可以通过Flink的数据处理操作来实现，例如使用union操作将多个数据流合并成一个。

4. 映射库名和表名: 在处理数据之前，你可能需要将源表中的库名和表名映射到目标表中的相应字段。这可以通过Flink的转换操作来实现，例如使用map或flatMap函数来提取和转换这些字段。

5. 写入目标表: 最后，将处理后的数据写入到目标表中。这个目标表应该包含库名和表名的信息，以便在查询时可以正确地引用源数据库中的表。

下面是一个示例代码片段，展示了如何在Flink中使用CDC Connector来捕获源表的变更数据，并将库名和表名信息同步到目标表中：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.DataTypes;

// 创建Flink执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 定义源表结构
tableEnv.executeSql("""
CREATE TABLE source_table (
    database STRING,
    table STRING,
    -- 其他字段
) WITH (
    'connector' = 'cdc',
    'hostname' = '<host>',
    'port' = '<port>',
    'username' = '<username>',
    'password' = '<password>',
    'database-name' = '<database-name>',
    -- 其他配置参数
)
""");

// 定义目标表结构
tableEnv.executeSql("""
CREATE TABLE sink_table (
    database STRING,
    table STRING,
    -- 其他字段
) WITH (
    'connector' = '<sink-connector>',
    -- 其他配置参数
)
""");

// 读取源表数据并写入目标表
tableEnv.executeSql("""
INSERT INTO sink_table
SELECT database, table, -- 其他字段
FROM source_table
""");

// 执行作业
env.execute("Flink CDC Job");

请注意，上述代码中的<host>、<port>、<username>、<password>、<database-name>以及<sink-connector>等参数需要根据实际情况进行替换。此外，你可能还需要根据具体的CDC Connector文档进行适当的配置和调整。

“MySQL CDC Source支持元数据列语法” 搜索
请参考此文档https://help.aliyun.com/zh/flink/developer-reference/mysql-connector?spm=a2c4g.11174283.0.i2
--此回答整理自钉群“实时计算Flink产品交流群”