mysql flink cdc kafka

### 使用 Flink CDC 实现 MySQL 数据变更捕获与 Kafka 流处理集成 #### 1. 添加依赖项 为了实现这一功能，在项目中需引入必要的 Maven 或 Gradle 依赖来支持 Flink、Flink CDC 和 Kafka 的连接器。 对于 Maven 用户来说，`pom.xml` 文件应包含如下片段： ```xml <dependencies> <!-- Flink dependencies --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- Flink CDC Connector for MySQL --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>2.3.0</version> </dependency> <!-- Kafka connector dependency --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>${flink.version}</version> </dependency> </dependencies> ``` #### 2. 创建源表定义 通过 SQL DDL 定义一个逻辑上的 "source table" 来表示来自 MySQL 的输入流。此操作会自动启动 Debezium 连接器监听指定数据库的变化日志。 ```sql CREATE TABLE mysql_source ( id BIGINT, name STRING, description STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', -- 替换为实际主机名 'port' = '3306', -- 默认端口可调整 'username' = 'root', -- 更改为有效用户名 'password' = 'your_password_here', -- 设置相应密码 'database-name' = 'test_db', -- 指定要监控的数据库名称 'table-name' = 'orders' -- 关注的具体表格列表 ); ``` #### 3. 构建 Sink 表至 Kafka 主题 同样利用 SQL 声明方式创建目标存储位置——即 Kafka topic ——用于接收由上述 source 所产生的记录序列。 ```sql CREATE TABLE kafka_sink ( id BIGINT, name STRING, description STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'order_updates_topic', -- 自定义主题名字 'properties.bootstrap.servers' = 'localhost:9092', -- Kafka broker 地址 'format' = 'json' ); ``` #### 4. 启动数据管道作业 最后一步是编写一段简单的 Java/Scala 应用程序代码用来提交整个 ETL 工作流程给 Apache Flink 集群执行。 ```java import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; // ... other imports ... public class MysqlToKafkaJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); String createSourceTableDDL = "..."; // 上述 Source Table DDL String createSinkTableDDL = "..."; // 上述 Sink Table DDL tEnv.executeSql(createSourceTableDDL); tEnv.executeSql(createSinkTableDDL); // 插入查询语句将变化的数据从 MySQL 发送到 Kafka 中去 tEnv.executeSql( "INSERT INTO kafka_sink SELECT * FROM mysql_source" ); env.execute("Mysql to Kafka Data Pipeline"); } } ``` 以上过程展示了如何配置 Flink CDC 以实现实时捕获 MySQL 数据库内的变动，并借助于 Apache Kafka 将其高效地分发出去[^1]。