Flink学习---15、FlinkCDC（CDC介绍、案例实操）

在这里插入图片描述
星光下的赶路人star的个人主页

未来总是藏在迷雾中让人胆怯，但当你踏入其中，便会云开雾散

文章目录

1、CDC简介
- 1.1 什么是CDC
- 1.2 CDC的种类
- 1.3 Flink-CDC
2、FlinkCDC案例实操
- 2.1 开启MySQL Binlog并重启MySQL
- 2.2 FlinkSQL方式的应用
- - 2.2.1 导入依赖
  - 2.2.2 编写代码

1、CDC简介

1.1 什么是CDC

CDC是Change Data Capture（变更数据获取）的简称。核心思想是，检测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件以及其他服务进行订阅及消费。

1.2 CDC的种类

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

	基于查询的CDC	基于Binlog的CDC
开源产品	Sqoop、DataX	Canal、Maxwell
执行模式	Batch	Streaming
是否可以捕获所有数据变化	否	是
延迟性	高延迟	低延迟
是否增加数据库压力	是	否

1.3 Flink-CDC

Flink摄取开发了Flink-cdc-connectors组件，这是一个可以直接从MySQL、PostgreSQL等数据库直接读取全量数据和增量变更数据的Source组件，
目前已经开源，开源地址：https://github.com/ververica/flink-cdc-connectors
在这里插入图片描述

2、FlinkCDC案例实操

2.1 开启MySQL Binlog并重启MySQL

vim /etc/my.cnf

在这里插入图片描述

2.2 FlinkSQL方式的应用

2.2.1 导入依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink-loader
</artifactId><version>1.17.0</version>
</dependency>

2.2.2 编写代码

public class FlinkCDC_SQL {public static void main(String[] args) throws Exception {// TODO 1. 准备环境// 1.1 流处理环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 1.2 表执行环境StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// TODO 2. 创建动态表tableEnv.executeSql("CREATE TABLE user_info (\n" +"id INT,\n" +"name STRING,\n" +"age INT,\n" +"primary key(id) not enforced\n" +") WITH (" +"'connector' = 'mysql-cdc'," +"'hostname' = 'hadoop102'," +"'port' = '3306'," +"'username' = 'root'," +"'password' = '123456'," +"'database-name' = 'test'," +"'table-name' = 'user'" +")");tableEnv.executeSql("select * from user_info").print();// TODO 3. 执行任务env.execute();}
}