是的，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，它提供了更好的集成和更简洁的配置方式。