在Flink CDC中mysql同步到doris，库名不同的情况，要怎么配置？

要实现MySQL同步到Doris，并且源库和目标库的库名不同，可以通过配置文件中的route部分来实现库名和表名的映射
定义源和目标表：在配置文件中，使用source部分定义MySQL数据库的连接信息和需要同步的表，使用sink部分定义Doris的连接信息。

路由配置：在route部分，通过source-table指定源表（包括库名和表名），通过sink-table指定目标表（包括目标库名和表名）。这样可以实现将不同库名的表同步到统一的库名下
大致就是这样

参考文档

在Flink CDC中配置MySQL同步到Doris，且源数据库（MySQL）与目标数据库（Doris）库名不同时，您需要调整命令行参数以适应这种情况。具体配置步骤如下：

1. 基本运行命令:
   首先，确保您的基本运行命令结构正确，根据给定示例，命令模板如下：

   <FLINK_HOME>/bin/flink run \
   -Dexecution.checkpointing.interval=10s \
   -Dparallelism.default=1 \
   -c org.apache.doris.flink.tools.cdc.CdcTools \
   lib/flink-doris-connector-1.16-1.5.2.jar \
   <mysql-sync-database> \
   --database <selectdb-database-name> \
   ...
   

2. 指定源数据库与目标数据库:

   • 对于源数据库（MySQL），Flink CDC会自动读取您指定的数据库（通过<mysql-sync-database>参数）。这里假设您的MySQL数据库名为source_db。
   • 对于目标数据库（Doris），使用--database参数指定同步到的Doris数据库名。如果源库名与目标库名不同，比如目标库名为target_db，则应配置为：

     --database target_db
     

3. 其他配置保持不变:
   继续按照需求配置其他参数，如--table-prefix、--table-suffix、--including-tables、--excluding-tables等，来精确控制哪些表被同步以及如何命名。

4. 确保依赖正确:
   确保$FLINK_HOME/lib目录下已添加了相应的Flink CDC依赖，例如flink-sql-connector-mysql-cdc-${version}.jar。

5. Sink配置:
   在Sink配置部分，通过fenodes、table.identifier、username、password等参数正确设置Doris实例的信息。特别是table.identifier，应按照Doris的库名和表名格式指定，例如：

   --sink-conf fenodes=selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080 \
   --sink-conf table.identifier=target_db.target_table
   

综上所述，在库名不同的场景下，主要通过调整--database参数以及Sink配置中的table.identifier来实现MySQL到Doris的库名映射。这样，即使源数据库与目标数据库的名称不同，也能顺利完成数据同步任务。

在Flink CDC中，当需要将MySQL同步到Doris，并且源MySQL数据库名与目标Doris数据库名不一致时，可以通过配置sink端的table.identifier参数来指定目标数据库和表名。以下是配置示例：

1. 配置sink端参数:

   • table.identifier: 此参数用于指定Doris中的数据库和表名。当MySQL的数据库名与Doris的目标数据库名不一致时，你需要在这里明确指定Doris的目标数据库名和表名。格式一般为<目标数据库名>.<表名>。

   例如，如果你的MySQL数据库名为mysql_db，表名为tbl1，而你想同步到Doris的数据库名为doris_db，则应配置如下：

   --sink-conf table.identifier=doris_db.tbl1
   

2. 命令行示例:
   假设其他配置保持不变，仅需调整数据库名，你的Flink命令行执行示例应调整为：

   <FLINK_HOME>/bin/flink run \
      -Dexecution.checkpointing.interval=10s \
      -Dparallelism.default=1 \
      -c org.apache.doris.flink.tools.cdc.CdcTools \
      lib/flink-doris-connector-1.16-1.5.2.jar \
      mysql-sync-database \
      --database mysql_db \ # MySQL源数据库名
      --mysql-conf hostname=127.0.0.1 \
      --mysql-conf username=root \
      --mysql-conf password=123456 \
      --mysql-conf database-name=mysql_db \
      --including-tables "tbl1|test.*" \
      --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
      --sink-conf username=admin \
      --sink-conf password=**** \
      --sink-conf table.identifier=doris_db.tbl1 # 指定Doris的目标数据库和表名
   

在使用 Flink CDC 将 MySQL 数据同步到 Doris 时，如果源 MySQL 数据库和目标 Doris 表所在的库名不同，可以通过 Flink CDC 的转换规则来实现这一需求。这里是一个简单的示例说明如何配置 Flink CDC 以支持这种场景。

步骤 1: 安装和配置 Flink CDC
首先确保你已经安装了 Flink CDC 并且配置了正确的 MySQL 和 Doris 连接信息。你可以参考 Flink CDC 的官方文档来完成这个过程。

步骤 2: 配置转换规则
假设你的 MySQL 中有一个名为 mysql_db 的数据库，而你想将数据同步到 Doris 的 doris_db 数据库中。你需要定义一个转换规则来更改表所在的库名。

配置文件示例
在 Flink CDC 的配置文件中，你可以使用 debezium 的转换规则来改变表名或者库名。这里是一个示例配置文件，展示了如何将 MySQL 中的库名 mysql_db 更改为 Doris 中的 doris_db。

配置Flink CDC从MySQL同步到Doris时，如果源数据库（MySQL）和目标数据库（Doris）的库名不同，您需要在Flink作业的配置中指定源和目标库名。以下是一个基本的配置示例：
指定MySQL的源库名：

指定Doris的目标库名：

确保将source_db_name替换为MySQL的源数据库名，将target_db_name替换为Doris的目标数据库名。同时您需要确保Flink CDC配置中的其他参数，如表名、用户名、密码、主机和端口等都是正确的。

指定源数据库名: 使用 --database 参数来指定MySQL的源数据库名。例如，如果MySQL的源数据库名为 mysql_db，则配置为：

指定目标Doris数据库名: 在sink配置 (sink-conf) 中，通过 database 参数设置Doris的目标数据库名。如果目标Doris数据库名为 test_db，则配置为：

参考文档

在Flink CDC中将MySQL的数据同步到Doris，且源库和目标库的库名不同，您需要在Flink作业的配置中指定源数据库和目标数据库的库名映射。以下是一个基本的配置示例：

# MySQL作为源数据库的配置
connector.type = mysql
connector.url = jdbc:mysql://<mysql_host>:<mysql_port>/<mysql_source_db>?useSSL=false
connector.username = <mysql_user>
connector.password = <mysql_password>
connector.table = <mysql_source_table>

# Doris作为目标数据库的配置，假设库名为doris_target_db
sink.type = flink-connector-doris
sink.doris.broker.address = <doris_broker_address>
sink.doris.username = <doris_user>
sink.doris.password = <doris_password>
sink.doris.db = doris_target_db
sink.doris.table = <doris_target_table>

在这个例子中，是源MySQL数据库的库名，是Doris的目标库名。您需要将这些占位符替换为实际的值，并确保在Flink作业的转换逻辑中处理库名的映射，或者在Doris中提前创建好对应名称的库。配置时需确保Flink CDC和Doris的连接参数正确，并且Doris的目标库存在。如果库不存在，您需要在Doris中先创建。

配置 Flink MySQL CDC Connector 来捕获 MySQL 数据库的变更。
定义源表和目标表之间的映射关系，包括库名和表名的映射。

flink.properties:

# 通用配置
jobmanager.rpc.address: <jobmanager-hostname>
taskmanager.numberOfTaskSlots: 2
blob.server.port: 6124
queryable-state.server.ports: 6125

# MySQL CDC Connector 配置
tables:
  - table-name: "source_database.source_table"
    schema:
      from: "source_database.source_table"
      to: "target_database.target_table"
    connection:
      hostname: "mysql_server_host"
      port: 3306
      user: "username"
      password: "password"
      database: "source_database"

# Doris Connector 配置
sinks:
  - table-name: "target_database.target_table"
    connection:
      feNodes: "doris_fe_node1:8030,doris_fe_node2:8030"
      username: "doris_username"
      password: "doris_password"
      database: "target_database"
    options:
      sink.buffer-flush.interval: 15
      sink.buffer-flush.max-rows: 500

在Flink CDC的配置文件中明确指定源数据库的库名和表名，以及在Doris中对应的表名。配置如下：

source:  
  type: mysql-cdc  
  properties:  
    hostname: your-mysql-host  
    port: 3306  
    username: your-mysql-username  
    password: your-mysql-password  
    database-name: mysql_source_db_name  # 源MySQL数据库的库名  
    table-name: your_table_name  # 或者使用正则表达式来匹配多个表，如 mysql_source_db_name\..*  
    server-id: 54321  # 确保这个ID在MySQL中唯一  

sink:  
  type: doris  
  properties:  
    fenodes: your-doris-fe-host:9030  # Doris FE的HTTP地址和端口  
    username: your-doris-username  
    password: your-doris-password  
    database-name: doris_target_db_name  # Doris数据库的库名，这里可以不同于MySQL的库名  
    table-name: your_target_table_name  # Doris中对应的表名，也需要你提前创建好  

pipeline:  
  name: mysql-to-doris  
  parallelism: 1  # 可以根据实际需要调整并行度

需要在Flink SQL中做一些配置来映射这些差异。以下是一个基本的步骤指南，帮助您配置同步过程：

添加MySQL CDC连接器依赖：确保您的Flink环境中已经添加了MySQL CDC连接器的依赖。
启动Flink SQL Client：使用Flink SQL Client来运行SQL语句。
创建MySQL CDC源表：在Flink中创建一个源表，用于从MySQL数据库读取数据。
创建Doris目标表：在Flink中创建一个目标表，用于将数据写入Doris。

Flink CDC 提供了将源表的表结构/数据路由到其他表名的配置，借助这种能力，我们能够实现表名库名替换，整库同步等功能。 下面提供一个配置文件说明：

参考文档https://nightlies.apache.org/flink/flink-cdc-docs-release-3.0/zh/docs/get-started/quickstart/mysql-to-doris/