大数据-业务数据采集-FlinkCDC

CDC

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

CDC 主要分为基于查询和基于 Binlog 两种方式，我们主要了解一下这两种之间的区别：
基于查询的 CDC基于 Binlog 的 CDC开源产品Sqoop、Kafka JDBC SourceCanal、Maxwell、Debezium执行模式BatchStreaming是否可以捕获所有数据变化否是延迟性高延迟低延迟是否增加数据库压力是否FlinkCDC

Flink 社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 MySQL、PostgreSQL 等数据库直接读取【全量数据】和【增量变更数据】的 source 组件。而不需要使用类似 Kafka 之类的中间件中转数据
目前也已开源，开源地址：https://github.com/ververica/flink-cdc-connectors


ConnectorDatabaseDrivermongodb-cdcMongoDB: 3.6, 4.x, 5.0MongoDB Driver: 4.3.1mysql-cdcMySQL: 5.6, 5.7, 8.0.x
RDS MySQL: 5.6, 5.7, 8.0.x
PolarDB MySQL: 5.6, 5.7, 8.0.x
Aurora MySQL: 5.6, 5.7, 8.0.x
MariaDB: 10.x
PolarDB X: 2.0.1JDBC Driver: 8.0.27oceanbase-cdcOceanBase CE: 3.1.x
OceanBase EE (MySQL mode): 2.x, 3.xJDBC Driver: 5.1.4xoracle-cdcOracle: 11, 12, 19Oracle Driver: 19.3.0.0postgres-cdcPostgreSQL: 9.6, 10, 11, 12JDBC Driver: 42.2.12sqlserver-cdcSqlserver: 2012, 2014, 2016, 2017, 2019JDBC Driver: 7.2.2.jre8tidb-cdcTiDB: 5.1.x, 5.2.x, 5.3.x, 5.4.x, 6.0.0JDBC Driver: 8.0.27db2-cdcDb2: 11.5DB2 Driver: 11.5.0.0DataStream:

• 优点： 多库多表
  
• 缺点： 需要自定义反序列化器（但灵活）
  FlinkSQL:
  
• 优点： 不需要自定义反序列化器
  
• 缺点： 单表
  

Demo

注意开启 binlog_format=ROW
my.ini

1. log-bin=mysql-bin
2. #binlog_format="STATEMENT"
3. binlog_format="ROW"
4. #binlog_format="MIXED"
5. #service-id=1

复制代码

POM

1.   <dependencies>
2.         <dependency>
3.             <groupId>org.apache.flink</groupId>
4.             <artifactId>flink-java</artifactId>
5.             <version>1.12.0</version>
6.         </dependency>
8.         <dependency>
9.             <groupId>org.apache.flink</groupId>
10.             <artifactId>flink-streaming-java_2.12</artifactId>
11.             <version>1.12.0</version>
12.         </dependency>
14.         <dependency>
15.             <groupId>org.apache.flink</groupId>
16.             <artifactId>flink-clients_2.12</artifactId>
17.             <version>1.12.0</version>
18.         </dependency>
20.         <dependency>
21.             <groupId>org.apache.hadoop</groupId>
22.             <artifactId>hadoop-client</artifactId>
23.             <version>3.1.3</version>
24.         </dependency>
26.         <dependency>
27.             <groupId>mysql</groupId>
28.             <artifactId>mysql-connector-java</artifactId>
29.             <version>5.1.49</version>
30.         </dependency>
32.         <dependency>
33.             <groupId>com.alibaba.ververica</groupId>
34.             <artifactId>flink-connector-mysql-cdc</artifactId>
35.             <version>1.2.0</version>
36.         </dependency>
38.         <dependency>
39.             <groupId>com.alibaba</groupId>
40.             <artifactId>fastjson</artifactId>
41.             <version>1.2.75</version>
42.         </dependency>
44.         <dependency>
45.             <groupId>org.apache.flink</groupId>
46.             <artifactId>flink-table-planner-blink_2.12</artifactId>
47.             <version>1.12.0</version>
48.         </dependency>
49.     </dependencies>

复制代码

基于 DataStream

CustomerDeserialization.java

1. package com.vipsoft;
3. import com.alibaba.fastjson.JSONObject;
4. import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
5. import io.debezium.data.Envelope;
6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
7. import org.apache.flink.api.common.typeinfo.TypeInformation;
8. import org.apache.flink.util.Collector;
9. import org.apache.kafka.connect.data.Field;
10. import org.apache.kafka.connect.data.Schema;
11. import org.apache.kafka.connect.data.Struct;
12. import org.apache.kafka.connect.source.SourceRecord;
14. import java.util.List;
16. public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
18.     /**
19.      * 封装的数据格式
20.      * {
21.      * "database":"",
22.      * "tableName":"",
23.      * "before":{"id":"","tm_name":""....},
24.      * "after":{"id":"","tm_name":""....},
25.      * "type":"c u d",
26.      * //"ts":156456135615
27.      * }
28.      */
29.     @Override
30.     public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
32.         //1.创建JSON对象用于存储最终数据
33.         JSONObject result = new JSONObject();
35.         //2.获取库名&表名
36.         String topic = sourceRecord.topic();
37.         String[] fields = topic.split("\\.");
38.         String database = fields[1];
39.         String tableName = fields[2];
41.         Struct value = (Struct) sourceRecord.value();
42.         //3.获取"before"数据
43.         Struct before = value.getStruct("before");
44.         JSONObject beforeJson = new JSONObject();
45.         if (before != null) {
46.             Schema beforeSchema = before.schema();
47.             List<Field> beforeFields = beforeSchema.fields();
48.             for (Field field : beforeFields) {
49.                 Object beforeValue = before.get(field);
50.                 beforeJson.put(field.name(), beforeValue);
51.             }
52.         }
54.         //4.获取"after"数据
55.         Struct after = value.getStruct("after");
56.         JSONObject afterJson = new JSONObject();
57.         if (after != null) {
58.             Schema afterSchema = after.schema();
59.             List<Field> afterFields = afterSchema.fields();
60.             for (Field field : afterFields) {
61.                 Object afterValue = after.get(field);
62.                 afterJson.put(field.name(), afterValue);
63.             }
64.         }
66.         //5.获取操作类型  CREATE UPDATE DELETE
67.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
68.         String type = operation.toString().toLowerCase();
69.         if ("create".equals(type)) {
70.             type = "insert";
71.         }
73.         //6.将字段写入JSON对象
74.         result.put("database", database);
75.         result.put("tableName", tableName);
76.         result.put("before", beforeJson);
77.         result.put("after", afterJson);
78.         result.put("type", type);
80.         //7.输出数据
81.         collector.collect(result.toJSONString());
83.     }
85.     @Override
86.     public TypeInformation<String> getProducedType() {
87.         return BasicTypeInfo.STRING_TYPE_INFO;
88.     }
89. }

复制代码

FlinkCDC.java

1. package com.vipsoft;
3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
5. import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
6. import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. public class FlinkCDC {
12.     public static void main(String[] args) throws Exception {
14.         //1.获取执行环境
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.         env.setParallelism(1);
18.         //2.通过FlinkCDC构建SourceFunction并读取数据
19.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
20.                 .hostname("localhost")
21.                 .serverTimeZone("GMT+8")  //时区报错增加这个设置
22.                 .port(3306)
23.                 .username("root")
24.                 .password("110")
25.                 .databaseList("springboot")
26.                 .tableList("springboot.sys_user")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
27.                 //.deserializer(new StringDebeziumDeserializationSchema())
28.                 .deserializer(new CustomerDeserialization()) //使用自定义反序列化器
29.                 .startupOptions(StartupOptions.initial())
30.                 .build();
31.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
33.         //3.打印数据
34.         streamSource.print();
36.         //4.启动任务
37.         env.execute("FlinkCDC");
39.     }
40. }

复制代码

运行效果

• 默认 StringDebeziumDeserializationSchema
  
  
• 自定义反序列化器
  
  

FlinkSQL

1. package com.vipsoft;
3. import org.apache.flink.api.java.tuple.Tuple2;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
6. import org.apache.flink.table.api.Table;
7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
8. import org.apache.flink.types.Row;
10. public class FlinkCDCWithSQL {
12.     public static void main(String[] args) throws Exception {
14.         //1.获取执行环境
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.         env.setParallelism(1);
17.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
19.         //2.DDL方式建表
20.         tableEnv.executeSql("CREATE TABLE mysql_binlog ( " +
21.                 " id STRING NOT NULL, " +
22.                 " username STRING, " +
23.                 " nick_name STRING " +
24.                 ") WITH ( " +
25.                 " 'connector' = 'mysql-cdc', " +
26.                 " 'hostname' = 'localhost', " +
27.                 " 'port' = '3306', " +
28.                 " 'username' = 'root', " +
29.                 " 'password' = '110', " +
30.                 " 'database-name' = 'springboot', " +
31.                 " 'table-name' = 'sys_user' " +
32.                 ")");
34.         //3.查询数据
35.         Table table = tableEnv.sqlQuery("select * from mysql_binlog");
37.         //4.将动态表转换为流
38.         DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);
39.         retractStream.print();
41.         //5.启动任务
42.         env.execute("FlinkCDCWithSQL");
43.     }
44. }

复制代码

运行效果


对比

通过对比，FlinkCDC 最舒服
FlinkCDCMaxwellCanal断点续传CKMySQL本地磁盘SQL -> 数据无无一对一（炸开处理）初始化功能有（多库多表）有（单表）无(单独查询历史数据)封装格式自定义JSONJSON（c/s自定义）高可用运行集群高可用无集群（ZK)插入对比

插入两条数据

1. INSER INTO z_user_info VALUES(30,'zhang3','13800000000'),(31,'li4','13999999999')

复制代码

FlinkCDC 每条变化都会产生一条 json

Maxwell 每条变化都会产生一条 json

Canal 一次性执行的SQL，会产生一条JSON（两条数据组合在一起）【不方便，需要炸开解析】

更新对比

1. UPDATE z_user_info SET user_name='wang5' WHERE id IN(30,31)

复制代码

FlinkCDC 包括了修改前的 before 数据

Maxwell 不包括修改前的数据

Canal 仍然是一条json

删除对比

1. DELETE FROM z_user_info WHERE id IN(30,31)

复制代码

FlinkCDC 两条删除的 json 数据

Maxwell

Canal

【尚硅谷】Flink数据仓库视频教程

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！