Flink CDC 实时监听 MySQL 表数据变化完整方案

Apache Flink CDC（Change Data Capture）可以实时捕获 MySQL 数据库中的数据更改，并将其推送到 Flink 流式处理引擎中。

以下是 FLINK CDC 监听 mysql 表数据变化的完整方案：

安装 Flink CDC 插件

先安装 Flink CDC 插件，使用以下命令：

flink-sql-connector-mysql-cdc_2.11-1.12.0.jar

创建 MySQL 数据库

在 MySQL 中创建一个数据库，例如 mytest，以及一个表，例如 user_info，用于测试。

配置 Flink 流式处理引擎

在 Flink 配置文件中，我们需要配置 MySQL 数据库的连接信息和 CDC 插件的相关参数。下面是一个示例配置文件：

flink.sources.mysql-1.type=mysql-cdc
flink.sources.mysql-1.url=jdbc:mysql://localhost:3306/mytest
flink.sources.mysql-1.username=root
flink.sources.mysql-1.password=123456
flink.sources.mysql-1.server-id=1
flink.sources.mysql-1.database-name=mytest
flink.sources.mysql-1.table-name=user_info
flink.sources.mysql-1.debezium.plugin.name=mysql

其中，flink.sources.mysql-1.type=mysql-cdc 表示使用 Flink CDC 插件进行 MySQL 数据捕获。flink.sources.mysql-1.url、flink.sources.mysql-1.username 和 flink.sources.mysql-1.password 是连接 MySQL 数据库所需的信息。flink.sources.mysql-1.server-id 是用于启用 MySQL 二进制日志的服务器 ID。flink.sources.mysql-1.database-name 和 flink.sources.mysql-1.table-name 是需要捕获的数据库和表名称。

编写 Flink 数据处理程序

在 Flink 中，我们可以使用 DataStream API 或 Table API 来处理数据。下面是一个使用 DataStream API 的示例程序：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties props = new Properties();
props.setProperty('bootstrap.servers', 'localhost:9092');
props.setProperty('group.id', 'flink-cdc');

DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<'mytest', new SimpleStringSchema(), props>);

stream.print();

env.execute();

在这个例子中，我们使用了 Flink Kafka 连接器来将数据从 Flink CDC 传输到 Kafka 中，然后使用 Flink Kafka 消费者来读取数据。

运行程序

将 Flink 程序打包成 JAR 文件并在命令行中运行：

flink run myprogram.jar

测试数据变化

在 MySQL 中插入、更新或删除 user_info 表中的数据，观察 Flink CDC 是否能够正确捕获数据变化，并将其推送到 Flink 流式处理引擎中。

以上就是 FLINK CDC 监听 mysql 表数据变化的完整方案。