Flink CDC（SQL Client）毗连 MySQL 数据库教程
Flink CDC（SQL Client）毗连 MySQL 数据库教程这篇文章将指导怎样使用 Flink CDC 毗连到 MySQL 数据库，并捕获数据变动。我们将逐步完成以下操作：
1. 检查 Binlog 是否启用
首先，您须要确保 MySQL 的 Binlog 功能已经启用，由于 Flink CDC 依赖于 Binlog 来捕获数据变动。
-- 检查 Binlog 是否启用
SHOW VARIABLES LIKE 'log_bin';
如果 log_bin 返回的不是 ON 大概一个文件名，那么 Binlog 没有启用。
2. 检查 Binlog 格式
-- 检查 Binlog 格式
SHOW VARIABLES LIKE 'binlog_format';
确保 binlog_format 是 ROW。如果不是，您须要修改配置以启用 Binlog 并设置正确的格式。
3. 开启 Binlog 并配置相关参数
如果 log_bin 的值为 OFF，这意味着 MySQL 的二进制日志（Binlog）功能没有开启。以下是开启 Binlog 并配置相关参数的步骤：
3.1 编辑 MySQL 配置文件
找到 MySQL 的配置文件 /etc/my.cn/。
在 部门添加或修改以下配置：
log_bin = mysql-bin
binlog_format = ROW
server_id = 1
[*]log_bin 设置 Binlog 的日志文件名前缀。
[*]binlog_format 设置为 ROW，这是 Flink CDC 所需的格式。
[*]server_id 设置为一个唯一的整数，用于标识 MySQL 服务器。
3.2 重启 MySQL 服务
生存配置文件后，重启 MySQL 服务以使更改生效。重启命令取决于您的操作体系：
[*] 对于 Linux/Unix：
sudo systemctl restart mysql
大概
sudo service mysql restart
4. 创建 CDC 用户
创建一个具有恰当权限的 MySQL 用户，以便 Flink CDC 可以毗连到 MySQL 数据库并监控数据变化：
CREATE USER 'flinkcdc'@'%' IDENTIFIED BY 'FlinkCDC_123456';
GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flinkcdc' IDENTIFIED BY 'FlinkCDC_123456';
FLUSH PRIVILEGES;
5. 创建 MySQL 表和插入示例数据
以下是一个名为 users 的表，包含 id（主键）、name（姓名）和 age（年事）三个字段的创建语句：
-- 创建数据库（如果不存在）
CREATE DATABASE IF NOT EXISTS cdc;
USE cdc;
-- 创建表
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
age INT NOT NULL
) ENGINE=InnoDB;
-- 插入示例数据
INSERT INTO users (name, age) VALUES ('Alice', 30);
INSERT INTO users (name, age) VALUES ('Bob', 25);
INSERT INTO users (name, age) VALUES ('Charlie', 35);
INSERT INTO users (name, age) VALUES ('David', 40);
INSERT INTO users (name, age) VALUES ('Eve', 22);
6. 下载 Flink CDC JAR 包
下载 Flink CDC JAR 包并放到 Flink 安装目录下的 lib 目录中。您可以从 Maven 中央堆栈下载 flink-sql-connector-mysql-cdc 2.3.0 版本：
Flink SQL Connector for MySQL CDC
请留意，官网提示下载的谁人包缺少一些依赖在执行时会报错，须要下载 20M 以上的 JAR 包。
Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: com.ververica.cdc.debezium.utils.ResolvedSchemaUtils
https://img-blog.csdnimg.cn/img_convert/64a30bad89d8f8c273688d4c7fb4afa2.png
7. 启动 Flink 集群
须要先启动 Flink 集群，否则背面会提示无法毗连： Could not execute SQL statement. Reason:java.net.ConnectException: 拒绝毗连
start-cluster.sh
8. 在 Flink SQL Client 中执行
打开 sql-client.sh 执行以下命令：
CREATE TABLE mysql_source (
id INT NOT NULL,
name STRING,
age INT,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '192.168.56.152',
'port' = '3306',
'username' = 'flinkcdc',
'password' = 'FlinkCDC_123456',
'database-name' = 'cdc',
'table-name' = 'users'
);
SELECT * FROM mysql_source;
https://img-blog.csdnimg.cn/img_convert/0fef638e25ddb9e8b28132e311642ed6.png
通过以上步骤，您应该能够成功使用 Flink CDC 毗连到 MySQL 数据库并捕获数据变动。如果在执行过程中遇到任何标题，请检查配置和网络设置，确保所有服务正常运行。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]