CDC是什么？有没有合适的技术方案？

CDC 是 Change Data Capture(变更数据获取)的简称。核心思想是，监测并捕获数据库的

变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下

来，写入到消息中间件中以供其他服务进行订阅及消费。

CDC 主要分为基于查询和基于 Binlog 两种方式，这两种之间的区别：

	基于查询的 CDC	基于 Binlog 的 CDC
开源技术	Datax、Sqoop、Kafka JDBC Source	Canal、Maxwell、Debezium、Flink CDC
执行模式	Batch	Streaming
延迟性	高延迟	低延迟
是否增加数据库压力	是	否

基于查询的 CDC 就不做介绍了，下文主要介绍一下 Flink 社区开源的 Flink CDC:

Flink 社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 MySQL、PostgreSQL

等数据库直接读取全量数据和增量变更数据的 source 组件。目前也已开源，开源地址：

https://github.com/ververica/flink-cdc-connectors

社区文档：CDC Connectors for Apache Flink® — CDC Connectors for Apache Flink® documentation

下面根据官方网站提供的教程，体验一下 Flink CDC

安装 Docker 环境

创建文件 docker-compose.yml

version: '2.1'
services:postgres:image: debezium/example-postgres:1.1ports:- "5432:5432"environment:- POSTGRES_PASSWORD=1234- POSTGRES_DB=postgres- POSTGRES_USER=postgres- POSTGRES_PASSWORD=postgresmysql:image: debezium/example-mysql:1.1ports:- "3306:3306"environment:- MYSQL_ROOT_PASSWORD=123456- MYSQL_USER=mysqluser- MYSQL_PASSWORD=mysqlpwelasticsearch:image: docker.elastic.co/elasticsearch/elasticsearch:7.10.2environment:- cluster.name=docker-cluster- bootstrap.memory_lock=true- "ES_JAVA_OPTS=-Xms512m -Xmx512m"- discovery.type=single-nodeports:- "9200:9200"- "9300:9300"ulimits:memlock:soft: -1hard: -1nofile:soft: 65536hard: 65536kibana:image: elastic/kibana:7.6.0ports:- "5601:5601"

使用命令，启动容器

docker-compose up -d

准备 Flink 环境

1.安装 Flink

访问页面 Index of /flink 这里选择 Flink 13.3

2.安装SQL连接器

下载下面列出的依赖包，并将它们放到目录 flink-1.13.3/lib/ 下：

flink-sql-connector-elasticsearch7_2.11-1.13.3.jar
flink-sql-connector-mysql-cdc-2.0.2.jar
flink-sql-connector-postgres-cdc-2.0.2.jar

3.准备数据

连接 mysql

docker-compose exec mysql mysql -uroot -p123456

创建表结构 & 数据

-- MySQL
CREATE DATABASE mydb;
USE mydb;
CREATE TABLE products (id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,name VARCHAR(255) NOT NULL,description VARCHAR(512)
);
ALTER TABLE products AUTO_INCREMENT = 101;INSERT INTO products
VALUES (default,"scooter","Small 2-wheel scooter"),(default,"car battery","12V car battery"),(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),(default,"hammer","12oz carpenter's hammer"),(default,"hammer","14oz carpenter's hammer"),(default,"hammer","16oz carpenter's hammer"),(default,"rocks","box of assorted rocks"),(default,"jacket","water resistent black wind breaker"),(default,"spare tire","24 inch spare tire");CREATE TABLE orders (order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,order_date DATETIME NOT NULL,customer_name VARCHAR(255) NOT NULL,price DECIMAL(10, 5) NOT NULL,product_id INTEGER NOT NULL,order_status BOOLEAN NOT NULL -- Whether order has been placed
) AUTO_INCREMENT = 10001;INSERT INTO orders
VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),(default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),(default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

连接 postgres

docker-compose exec postgres psql -h localhost -U postgres

创建表结构 & 数据