flink cdc 里高版本的 mysql cdc 取消掉这种创建数据源的方式了吗？ -凯发旗舰厅

是的，flink cdc 高版本取消了创建数据源的方式，改为使用 table api 或 sql api 来操作数据源。

flink cdc 中高版本 mysql cdc 取消创建数据源的方式

单元表格：

在 flink cdc（change data capture）中，高版本的 mysql cdc（mysql change data capture）取消了使用 debeziumsourcefunction 创建数据源的方式，取而代之的是，使用 debeziumdeserializationschema。

具体来说，旧版本中，我们可以通过实现 debeziumsourcefunction 来创建数据源，并在该函数中进行连接配置、初始化和关闭等操作，而在新版本中，这些操作需要在 flink sql ddl（data definition language）中进行配置和执行。

以下是使用新版本的步骤：

1、定义表结构：我们需要在 flink sql ddl 中定义要使用的表结构，这包括表名、字段名、字段类型等信息。

```sql

create table my_table (

id bigint,

name string,

age int,

...

) with (...);

```

2、配置数据源连接：接下来，我们需要在 flink sql ddl 中配置数据源的连接信息，这包括数据库 url、用户名、密码等。

```sql

set 'debezium.connector.class' = 'io.debezium.connector.mysql.mysqlconnector';

set 'debezium.offset.storage' = 'org.apache.flink.connector.debezium.offsetbackingstore';

set 'debezium.offset.storage.file.filename' = '/path/to/offset/storage/file';

set 'debezium.database.hostname' = 'localhost';

set 'debezium.database.port' = '3306';

set 'debezium.database.user' = 'root';

set 'debezium.database.password' = 'password';

set 'debezium.database.server.id' = '85740';

set 'debezium.database.server.name' = 'my_server';

set 'debezium.database.whitelist' = 'my_db,other_db';

```

3、初始化数据源：我们可以使用 flink sql ddl 中的其他语句来初始化数据源，可以使用 create table as select 语句将已有的数据导入到新表中。

```sql

create table my_table_copy as select * from my_source_table;

```

通过以上步骤，我们可以在新版本的 flink cdc 中使用 debeziumdeserializationschema 来创建和管理数据源，这种方式更加简洁和灵活，并且与 flink sql ddl 集成得更好。

相关问题与解答：

1、q: 新版本的 flink cdc 中如何关闭数据源？

a: 在新版本的 flink cdc 中，关闭数据源的操作需要在 flink sql ddl 中进行，可以使用 drop table 语句来删除对应的表，从而关闭数据源。drop table my_table;，这将释放相关资源并关闭数据源。

2、q: 我可以在新版本的 flink cdc 中使用旧版本的 debeziumsourcefunction 吗？

a: 不可以，在新版本的 flink cdc 中，取消了对旧版本的 debeziumsourcefunction 的支持，建议使用新版本提供的 debeziumdeserializationschema，它提供了更好的集成和更简洁的配置方式。