Flink_CDC搭建及简单使用

Flink_CDC搭建及简单使用
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.0.jar
 
mysql版本:8.0.13
 
mysql驱动包:mysql-connector-java-8.0.13.jar
 
3.2环境搭建
 
3.2.1安装java环境(不再赘述);
 
3.2.2安装数据库(不在赘述);
 
3.2.3搭建Flink环境(单机模式);
 
1.获取flink版本。
 
  cd /home
 
  wget https://archive.apache.org/dist/flink/flink-1.9.0/flink-1.13.1-bin-scala_2.11.tgz
 
2.解压flink:
 
 tar -zxvf flink-1.13.1-bin-scala_2.11.tgz 
 
3.编辑flink配置文件,配置java环境
 
cd flink-1.13.1
 
vim conf/flink-conf.yaml
 
添加配置:env.java.home=/home/jdk/jdk1.8.0_291
 
4.上传flink_CDC驱动包和mysql驱动包:
 
cd flink-1.13.1/lib
 
上传:
 
flink-sql-connector-mysql-cdc-2.1.0.jar
 
mysql-connector-java-8.0.13.jar
 
5.启动flink集群:
 
/bin/start-cluster.sh
 
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;
 
3.4启动flink-sql-client
 
​ 本次主要是通过flink的sql客户端来测试的。
 
./flink-1.13.1/bin/sql-client.sh
 
1
 
3.5创建Flink_CDC虚拟表:
 
CREATE TABLE ——products_cdc—— (
 
 id INT NOT NULL,
 
 name varchar(32),
 
 description varchar(45),
 
 weight DECIMAL(10,3)
 
) WITH (
 
 'scan.incremental.snapshot.enabled' = 'false',
 
 'connector' = 'mysql-cdc',
 
 'hostname' = '0.0.0.0',
 
 'port' = '3306',
 
 'username' = 'root',
 
 'password' = '123456',
 
 'database-name' = 'test_db',
 
 'table-name' = 'products'
 
);
 
###如果未设置'scan.incremental.snapshot.enabled' = 'false',会报错:
 
Caused by: org.apache.flink.table.api.ValidationException: The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled' , default: true (fallback keys: [])' to 'true'
 
    at com.alibaba.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.validatePrimaryKeyIfEnableParallel(MySqlTableSourceFactory.java:186)
 
    at com.alibaba.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.createDynamicTableSource(MySqlTableSourceFactory.java:85)
 
    at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:134)
 
    … 30 more
 
###报错原因:
 
  MySQL CDC源表在Flink 1.13版本会进行语法检查,在MySQL CDC DDL WITH参数中,未设置主键(Primary Key)信息。因为Flink 1.13版本,新增支持按PK分片,进行多并发读取数据的功能。
 
###解决方案:
 
如果在Flink 1.13版本,您需要多并发读取MySQL数据,则在DDL中添加PK信息。

推荐阅读