环境说明：
flink 1.15.2
mysql 版本5.7 注意：需要开启binlog，因为增量同步是基于binlog捕获数据
windows11 IDEA 本地运行
先上官网使用说明和案例：MySQL CDC Connector — Flink CDC documentation
1. mysql开启binlog (注意，引擎是 InnoDB,如果是ndbcluster，本人测试是捕获不到binlog日志的，增量相当于没用，不知道是不是ndbcluster 下的binlog 配置是否有问题，但是同一集群下，InnoDB的表就可以捕获到binlog日志。听朋友说，ndbcluster 是内存型引擎，有可能不会实时写日志到磁盘，所以捕获不到.....)
# 判断MySQL是否已经开启binlog on 为打开状态
SHOW VARIABLES LIKE 'log_bin';# 查看MySQL的binlog模式
show global variables like "binlog%";# 查看日志开启状态
show variables like 'log_%';# 刷新log日志，立刻产生一个新编号的binlog日志文件，跟重启一个效果
flush logs;# 清空所有binlog日志
reset master;
2. 创建一个用户，赋权
CREATE USER 'flink_cdc_user'@'%' IDENTIFIED BY 'flink@cdc';
GRANT ALL PRIVILEGES ON *.* TO 'flink_cdc_user'@'%';
3. maven依赖:
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<flink.version>1.15.2</flink.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_2.12</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
<!-- 此标签会移除jar包，当需要打包到集群运行时加上此标签-->
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.29</version>
<!-- <scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
</dependency>
<!-- https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-mysql-cdc -->
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-sql-connector-mysql-cdc</artifactId>
<version>2.3.0</version>
<!--<scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc</artifactId>
<version>1.15.2</version>
<!--<scope>provided</scope>-->
<!--此标签会移除jar包，当需要打包到集群运行时加上此标签-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-base</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>
</dependencies>4. 若是打包到集群运行，相关依赖要放开 provided，这样就不会把依赖打入到jar包里面，就不会和flink lib里面的jar包冲突。
lib 里面需要加入的包：从官网下载，放入即可
flink-connector-jdbc-1.15.4.jar
flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar
flink-sql-connector-mysql-cdc-2.3.0.jar
mysql-connector-java-8.0.29.jar
commons-cli-1.5.0.jar
5.mysql建表如下：
#mysql建表：
CREATE TABLE `user` (
`id` int(11) NOT NULL,
`username` varchar(255) DEFAULT NULL,
`password` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;CREATE TABLE `user_sink` (
`id` int(11) NOT NULL,
`username` varchar(255) DEFAULT NULL,
`password` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
6.测试demo如下：
package com.xgg.flink.stream.sql;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class MysqlToMysqlHavePrimaryKey {
public static void main(String[] args) {
//1.获取stream的执行环境
StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
senv.setParallelism(1);
//2.创建表执行环境
StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);
String sourceTable = "CREATE TABLE mysql_cdc_source (" +
" id INT,\n" +
" username STRING,\n" +
" password STRING,\n" +
"PRIMARY KEY(id) NOT ENFORCED\n" +
") WITH (\n" +
"'connector' = 'mysql-cdc',\n" +
"'hostname' = 'localhost',\n" +
"'port' = '3306',\n" +
"'username' = 'root',\n" +
"'password' = 'root',\n" +
"'database-name' = 'test_cdc',\n" +
"'debezium.snapshot.mode' = 'initial',\n" +
"'table-name' = 'user'\n" +
")";
tEnv.executeSql(sourceTable);
String sinkTable = "CREATE TABLE mysql_cdc_sink (" +
" id INT,\n" +
" username STRING,\n" +
" password STRING,\n" +
"PRIMARY KEY(id) NOT ENFORCED\n" +
") WITH (\n" +
"'connector' = 'jdbc',\n" +
"'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
"'url' = 'jdbc:mysql://localhost:3306/test_cdc?rewriteBatchedStatements=true',\n" +
"'username' = 'root',\n" +
"'password' = 'root',\n" +
"'table-name' = 'user_sink'\n" +
")";
tEnv.executeSql(sinkTable);
tEnv.executeSql("insert into mysql_cdc_sink select id,username,password from mysql_cdc_source");
tEnv.executeSql("select * from mysql_cdc_source").print();
}
}
源表进行操作，flink cdc 捕获操作记录进行打印，然后插入到表中。(mysql的cdc可以一边打印，一边写表，无问题。oracle的cdc，如果有多个执行操作，就会只执行一个，比如，先打印再写表，oracle只能打印，写表操作就不会触发。如果不打印，只写表，那就没问题。好像和senv.setParallelism(1);没关系，应该还是底层实现的问题。)
user 源表和目标表 user_sink,数据都如下。
文章来源：https://www.toymoban.com/news/detail-691039.html
源表和目标表都是在Mysql有主键的，所以找个参数虽然是初始化操作，后面插入也是 insert into ，但是不管执行多少遍，都不会有重复的数据。文章来源地址https://www.toymoban.com/news/detail-691039.html
"'debezium.snapshot.mode' = 'initial',\n" +
?rewriteBatchedStatements=true 这个参数是开启批量写，能加大写速度。
到了这里，关于Flink CDC 基于mysql binlog 实时同步mysql表的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
