Flink SQL Client实现CDC实验

概述

本文主要是對[7]中內容的復現

环境

组件	版本
Flink(HA)	1.12
Zookeeper	3.6.0
flink-sql-connector-mysql-cdc	1.1.1
Mysql	8.0.22-0ubuntu0.20.04.2

實驗流程圖

Mysql的同步配置

/etc/mysql/mysql.conf.d/mysqld.cnf中的[mysqld]下面添加:

# 前面还有其他配置
# 添加的部分
server-id = 12345
log-bin = mysql-bin
# 必须为ROW
binlog_format = ROW
# 必须为FULL，MySQL-5.7后才有该参数
binlog_row_image  = FULL
expire_logs_days  = 10

查看binlog默认配置:

SHOW VARIABLES LIKE '%binlog%';

service mysql restart

创建用于同步的用户，并给予权限（可供参考的文档）

-- 设置拥有同步权限的用户
CREATE USER 'appleyuchi' IDENTIFIED BY 'appleyuchi';
-- 赋予同步相关权限
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'appleyuchi';

创建用户并赋予权限成功后，使用该用户登录MySQL，可以使用以下命令查看主从同步相关信息

SHOW MASTER STATUS;
SHOW SLAVE STATUS;
SHOW BINARY LOGS;

详细操作步骤

操作內容	效果
Mysql建立表格
Mysql插入數據
Flink SQL Client建立Source
Flink SQL Client建立Sink
mysql_binlog的數據插入到tb_sink
查看當前Flink集羣的taskmanager中的數據
更新Mysql以後，再次查看Flink集羣的taskmanager中的數據

上述表格中各个步骤需要的SQL汇总如下:

https://gitee.com/appleyuchi/Flink_Code/blob/master/FLINK读写各种数据源/cdc.sql

這裏需要注意，如果本實驗的source中的mysql的IP是一個外網IP,那麼需要確保mysql所在節點可以被外網訪問，

否則會無法順利提交任務到集羣．上述鏈接中使用的localhost,沒有該問題

實驗結論

當Mysql中修改數據以後，我們會發現Flink集羣任務中的Task Manager的Stdout也會立刻做出修改．

因此CDC的同步功能順利實現

CDC其实就是类似于一个金山同步盘一样的功能，上游的数据改动后，同步到下游．

依赖问题

https://maven.aliyun.com/mvn/search

搜索flink-sql-connector-mysql-cdc-1.1.1.jar

然后放到$FLINK_HOME/lib下面，然后同步到集群的其他节点

異常

Caused by: java.sql.SQLNonTransientConnectionException: Public Key Retrieval is not allowed

如果集羣兩個節點Desktop和Laptop

此時Laptop無法登錄Desktop的mysql就會導致Flink集羣發生這種異常報錯

此时下述两个解决办法取其一即可:

①让Desktop中的mysql支持外网访问

②source中的mysql的IP从原本的Desktop改成localhost

Reference:

[1]Flink SQL Client + Mysql CDC 部署实践(yaml格式)

[2]关于flink:Flink-SQL-Client-Mysql-CDC-部署实践(yaml格式)

[3]Flink SQL CDC(涉及ElasticSearch的)

[4]Flink SQL CDC 上线！我们总结了 13 条生产实践经验(提到维度表Join和双流Join不一样)

[5]Flink1.11中的CDC Connectors操作实践(非常详细)

[6]基于 Flink SQL CDC的实时数据同步方案(没啥用,讲了一系列的场景和架构)

[7]Flink示例——Flink-CDC(DDL嵌入到代码中的形式)

[8]Mysql binlog详解(详细介绍)