【Flink-cdc-Mysql-To-Kafka】使用 Flinksql 利用集成的 connector 实现 Mysql 数据写入 Kafka

1）环境准备

Linux 或者 Windows 端需要安装：Mysql，Kafka，Flink 等。（略）

2）准备相关 jar 包

• flink-connector-jdbc_2.11-1.12.0.jar
• mysql-connector-java-5.1.49.jar

下载地址：JDBC-Sql-Connector

• flink-format-changelog-json-1.2.0.jar
• flink-sql-connector-mysql-cdc-1.2.0.jar
• flink-sql-connector-postgres-cdc-1.2.0.jar

下载地址：ververica/flink-cdc-connectors

备用下载地址：gitee地址(github上不去就下载源码,改好version自己打包)

• flink-sql-connector-kafka_2.11-1.12.0.jar

下载地址：flink-sql-connector-kafka

• 将下载好的包放在 Flink 的 lib 目录下

3）实现场景

1、首先确认MySQL是否开启binlog机制，log_bin = ON 为开启 （如下图）

2、如果是本地环境的 Mysql 按照下面方式开启 binlog

在 C:\ProgramData\MySQL\MySQL Server 5.7\my.ini 下添加

log_bin = mysql-bin
binlog_format = ROW
expire_logs_days = 30

3、重启 Mysql 服务

4）准备工作

4.1.Mysql

1、在 Mysql 中创建 source 表：

CREATE TABLE `mysql2kafka_cdc_test` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `eventId` varchar(255) DEFAULT NULL,
  `eventStDt` varchar(255) DEFAULT NULL,
  `bak6` varchar(255) DEFAULT NULL,
  `bak7` varchar(255) DEFAULT NULL,
  `businessId` varchar(255) DEFAULT NULL,
  `phone` varchar(255) DEFAULT NULL,
  `bak1` varchar(255) DEFAULT NULL,
  `bak2` varchar(255) DEFAULT NULL,
  `bak13` varchar(255) DEFAULT NULL,
  `bak14` varchar(255) DEFAULT NULL,
  `bak11` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8

2、写入数据的语句准备就绪

INSERT INTO mysql2kafka_cdc_test(
eventId,
eventStDt,
bak6,
bak7,
businessId,
phone,
bak1,
bak2,
bak13,
bak14,
bak11
) VALUES(
'111',
'2022-11-3023:37:49',
'测试',
'https://test?user',
'1727980911111111111111111111',
'12345678910',
'1234',
'2021-12-0100:00:00',
'1727980911111111111111111111',
'APP',
'TEST1'
);

4.2.Kafka

创建 Topic

5）Flink-Sql

• source

set table.dynamic-table-options.enabled=true;
set table.exec.source.cdc-events-duplicate=true;

CREATE TABLE source_mysql_test(
  id INT,
  eventId STRING,
  eventStDt STRING,
  bak6 STRING,
  bak7 STRING,
  businessId STRING,
  phone STRING,
  bak1 STRING,
  bak2 STRING,
  bak13 STRING,
  bak14 STRING,
  bak11 STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH(
    'connector' = 'mysql-cdc',
    'hostname' = '${ip}',
    'port' = '${port}',
    'database-name' = 'test',
    'table-name' = 'mysql2kafka_cdc_test',
    'username' = '${username}',
    'password' = '${password}',
    'scan.startup.mode'='timestamp',
    'scan.startup.timestamp-millis' = '1692115200000'
);

• sink

CREATE TABLE sink_kafka_test (
  id INT,
  eventId STRING,
  eventStDt STRING,
  bak6 STRING,
  bak7 STRING,
  businessId STRING,
  phone STRING,
  bak1 STRING,
  bak2 STRING,
  bak13 STRING,
  bak14 STRING,
  bak11 STRING,
  PRIMARY KEY (id) NOT ENFORCED
 ) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'test',
    'sink.parallelism' = '3',
    'key.format' = 'json',
    'value.format' = 'json',
    'properties.bootstrap.servers' = '${kafka-bootstrap-server}',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.kerberos.service.name' = 'kafka',
    'metadata.max.age.ms' = '300000'
);

• insert

insert into sink_kafka_test select * from source_mysql_test;

6）验证

Mysql 中写入测试数据，Kafka-Topic 中观察是否有数据生成。

INSERT INTO mysql2kafka_cdc_test(
eventId,
eventStDt,
bak6,
bak7,
businessId,
phone,
bak1,
bak2,
bak13,
bak14,
bak11
) VALUES(
'111',
'2022-11-3023:37:49',
'测试',
'https://test?user',
'1727980911111111111111111111',
'12345678910',
'1234',
'2021-12-0100:00:00',
'1727980911111111111111111111',
'APP',
'TEST1'
);