如何操作flink cdc 实时同步mysql

Flink CDC 是 Flink 社区针对 MySQL 数据库实现的一种 Change Data Capture（数据变更捕捉）解决方案。它可以将 MySQL 的 binlog 转化为 Flink 的 DataStream，从而实现实时同步 MySQL 数据库的功能。

下面是操作步骤：

1. 首先，需要在 MySQL 数据库中开启 binlog，以便 Flink CDC 可以通过 binlog 实现数据的实时同步。在 MySQL 的配置文件 my.cnf 中添加以下配置：

[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server_id=1

2. 在 Flink 中添加 flink-connector-mysql 依赖。可以通过 Maven 或 Gradle 将其添加到项目中。

3. 在 Flink 代码中编写 CDC 任务。以下是一个简单的示例：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// 创建 MySQL Source
DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
    .hostname("localhost")
    .port(3306)
    .username("root")
    .password("root")
    .databaseList("test")
    .tableList("test.user")
    .deserializer(new StringDeserializer())
    .build();

// 添加 Source
DataStream<String> stream = env.addSource(sourceFunction);

// 打印输出
stream.print();

// 执行任务
env.execute("Flink CDC Task");

在上面的示例中，我们创建了一个 MySQL Source，指定了 MySQL 数据库的连接信息、要同步的数据库和表，以及数据反序列化器。然后，我们将 Source 添加到 Flink 中，并通过 print() 方法将数据输出到控制台。最后，我们执行 Flink 任务。

4. 运行 Flink 任务。运行前，需要确保 MySQL 数据库已经开启 binlog，并且 Flink 代码中的连接信息正确。

以上就是操作 Flink CDC 实时同步 MySQL 的基本步骤。当然，实际情况可能更加复杂，需要根据具体场景进行调整。