Flink_CDC搭建及简单使用

时间：2024-10-07 11:01:57

Flink_CDC搭建及简单使用

1.CDC简介：

CDC （Change Data Capture） ，在广义的概念上，只要能捕获数据变更的技术，都可以称为 CDC 。但通常我们说的CDC 技术主要面向数据库（包括常见的mysql,Oracle, MongoDB等）的变更，是一种用于捕获数据库中数据变更的技术。

目前市面上的CDC技术非常多,常见的主要包括Flink CDC,DataX,Canal,Sqoop,Kettle,Oracle Goldengate,Debezium等。DataX，Sqoop和kettle的CDC实现技术主要是基于查询的方式实现的，通过离线调度查询作业，实现批处理请求。这种作业方式无法保证数据的一致性，实时性也较差。Flink CDC，Canal，Debezium和Oracle Goldengate是基于日志的CDC技术。这种技术，利用流处理的方式，实时处理日志数据，保证了数据的一致性，为其他服务提供了实时数据。

2.Flink_CDC简介：

目前公司主要是通过canal监控mysql的binlog日志，然后将日志数据实时发送到kafka中，通过flink程序，将日志数据实时下发到其他服务中。这种方式，数据链路长，实时性效果较差，运维也比较复杂。

Flink_CDC技术的出现，解决了传统数据库实时同步的痛点。Flink_CDC通过伪装成msql的slave节点，实时读取master节点全量和增量数据，它能够捕获所有数据的变化，捕获完整的变更记录，无需像查询CDC那样发起全表的扫描过滤，高效且无需入侵代码，完全与业务解耦，运维及其简单。

3.Flink_CDC部署：
3.1 依赖版本
环境：Linux(Centos7)
Flink : 1.31.1
Flink_CDC: flink-sql-connector-mysql-cdc-2.1.
mysql版本：8.0.13
mysql驱动包：mysql-connector-java-8.0.
  • 1
  • 2
  • 3
  • 4
  • 5
3.2环境搭建

3.2.1安装java环境(不再赘述)；

3.2.2安装数据库(不在赘述)；

3.2.3搭建Flink环境(单机模式);

1.获取flink版本。

  cd /home
  wget /dist/flink/flink-1.9.0/flink-1.13.1-bin-scala_2.
  • 1
  • 2

2.解压flink：

 tar -zxvf flink-1.13.1-bin-scala_2.
  • 1

3.编辑flink配置文件，配置java环境

cd flink-1.13.1
vim conf/
添加配置：=/home/jdk/jdk1.8.0_291
  • 1
  • 2
  • 3

4.上传flink_CDC驱动包和mysql驱动包：

cd flink-1.13.1/lib
上传：
flink-sql-connector-mysql-cdc-2.1.
mysql-connector-java-8.0.
  • 1
  • 2
  • 3
  • 4

在这里插入图片描述

5.启动flink集群：

/bin/
  • 1
3.3创建mysql表：
CREATE TABLE `products` (
   `id` int NOT NULL,
   `name` varchar(45) DEFAULT NULL,
   `description` varchar(45) DEFAULT NULL,
   `weight` decimal(10,3) DEFAULT NULL,
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
3.4启动flink-sql-client

本次主要是通过flink的sql客户端来测试的。

./flink-1.13.1/bin/
  • 1
3.5创建Flink_CDC虚拟表：
CREATE TABLE `products_cdc` (
 id INT NOT NULL,
 name varchar(32),
 description varchar(45),
 weight DECIMAL(10,3)
) WITH (
 '' = 'false',
 'connector' = 'mysql-cdc',
 'hostname' = '0.0.0.0',
 'port' = '3306',
 'username' = 'root',
 'password' = '123456',
 'database-name' = 'test_db',
 'table-name' = 'products'
);


###如果未设置'' = 'false'，会报错：

Caused by: : The primary key is necessary when enable 'Key: '' , default: true (fallback keys: [])' to 'true'
    at (:186)
    at (:85)
    at (:134)
    ... 30 more
    
###报错原因：

  MySQL CDC源表在Flink 1.13版本会进行语法检查，在MySQL CDC DDL WITH参数中，未设置主键（Primary Key）信息。因为Flink 1.13版本，新增支持按PK分片，进行多并发读取数据的功能。
  
###解决方案：

如果在Flink 1.13版本，您需要多并发读取MySQL数据，则在DDL中添加PK信息。
如果在Flink 1.13版本，您不需要多并发读取MySQL数据，则在DDL中添加 参数，且把该参数值设置为false，无需设置PK信息。
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
3.6查询CDC表数据：

3.6.1 查看表数据
select * from products_cdc;
在这里插入图片描述
3.6.2 在数据库中新增一条数据：
insert into products(id,name,description,weight) values(5,‘gg’,‘haha’,60);
在这里插入图片描述
3.6.3观察products_cdc表数据变化；
在这里插入图片描述

到此，通过flink-sql-client来增量获取mysql全量和增量数据变化的样例已结束。

最后

深知大多数初中级Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则近万的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《Java开发全套学习资料》送给大家，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

小编已加密：aHR0cHM6Ly9kb2NzLnFxLmNvbS9kb2MvRFVrVm9aSGxQZUVsTlkwUnc==出于安全原因，我们把网站通过base64编码了，大家可以通过base64解码把网址获取下来。

相关文章