请问Flinkcdc中整库同步都是不同表结构的吗？有示例吗？

在 Flink CDC 中，整库同步可以用于将 MySQL 数据库中的所有表的数据同步到 Flink 中。如果这些表的结构不同，那么 Flink CDC 应该如何处理呢？
Flink CDC 提供了 Schema Registry 功能，可以根据表名和表结构自动创建对应的 TableSchema，并将其注册到 Flink 中。在整库同步时，Flink CDC 会根据注册的 TableSchema 自动创建对应的 Flink 表，并将 MySQL 中的数据转换为 Flink 表中的数据。
以下是一个整库同步的示例，假设 MySQL 数据库中有三个表，分别为 table1、table2、table3，它们的结构如下：
mysql
Copy
CREATE TABLE table1 (
id INT PRIMARY KEY,
name VARCHAR(20)
);

CREATE TABLE table2 (
id INT PRIMARY KEY,
age INT
);

CREATE TABLE table3 (
id INT PRIMARY KEY,
address VARCHAR(50),
phone VARCHAR(20)
);
在 Flink CDC 中，可以使用以下代码将这些表的数据同步到 Flink 中：
java
Copy
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

tableEnv
.connect(new MySqlCdcTableSource(
"jdbc:mysql://localhost:3306/test",
"root",
"password",
"table1,table2,table3"))
.withFormat(new Json())
.withSchema(
new Schema()
.field("id", DataTypes.INT())
.field("name", DataTypes.STRING())
.field("age", DataTypes.INT())
.field("address", DataTypes.STRING())
.field("phone", DataTypes.STRING())
)
.createTemporaryTable("my_table");

tableEnv.executeSql("SELECT * FROM my_table").print();
在上述代码中，MySqlCdcTableSource 是 Flink CDC 提供的 MySQL CDC 数据源

Flink CDC 可以自动同步整库的全量和增量数据，同时还能实时将每张源表的表结构变更(加列等)实时同步到对应的目标表中。

以下是一个示例代码，可以将不同表结构的表进行整库同步：

TableEnvironment tEnv = TableEnvironment.create(environment);
// 读取源数据库中的表
SourceFunction<Row> source = CDCSource.<Row>builder()
        .setStartupOptions(startupOptions)
        .setConnectionOptions(connectionOptions)
        .setSql("SELECT * FROM my_table")
        .build();
// 将数据写入目标数据库中的表
StreamTableSink<Row> sink = StreamTableSink.forConnector(
        new FlinkDynamicTableSinkBase<>(sinkProperties), environment);
tEnv.executeSql("CREATE TABLE target_table AS SELECT * FROM source_table").execute().getJobClient().getJobExecutionResult().get();
tEnv.executeSql("INSERT INTO target_table SELECT * FROM source_table").execute().getJobClient().getJobExecutionResult().get();