声明: 1. 本文为我的个人复习总结, 并非那种从零基础开始普及知识 内容详细全面, 言辞官方的文章
2. 由于是个人总结, 所以用最精简的话语来写文章
3. 若有错误不当之处, 请指出
Change Data Capture 变更数据获取, 去MySQL等后端数据库拉取变更数据
核心思想:监测并捕获MySQL里数据的变动, 将这些变更按发生的顺序完整记录下来, 然后写入到Kafka中, 以供其他服务进行订阅及消费
原理:Canal, Maxwell, Flink-CDC是伪装成slave去主从复制拉数据, 但不具备sql执行引擎, 故采用的是BinLog的ROW模式
各种CDC:CDC主要分为基于查询和基于BinLog两种方式:
| 基于查询 | 基于BinLog |
|---|---|
| Sqoop | Canal、Maxwell、Debezium |
| 离线批处理 | 实时流处理 |
| 不能得到数据更新的历史记录, 只能得到最终结果 | 能得到数据更新的历史记录 |
| 高延迟 | 低延迟 |
| 增加了数据库压力(要执行select语句) | 没增加数据库压力 |
优点: 可以用Java程序作客户端去订阅Canal获取里面的数据
缺点: Canal只能捕获 在BinLog里有记录的 变更数据, 不能捕获 开启BinLog之前 就已经存在的历史数据
Maxwell:优点: Maxwell不仅能捕获 在BinLog里有记录的 变更数据, 而且还有一个bootstrap功能去捕获在 开启BinLog之前 就已经存在的历史数据
maxwell-bootstrap的底层还是maxwell进程
缺点: 不能用Java程序作客户端去订阅Maxwell;
要是想获取里面的数据, 只能在Maxwell配置文件里配置将数据保存到Kafka, 然后用Java程序去消费Kafka, 间接获取Maxwell里的数据
FlinkCDC:功能最强大:
- 既能捕获 开启BinLog之前 就已经存在的历史数据又因为本身是Flink-Java程序, 所以不必去订阅Flink-CDC, Java程序客户端也能直接获取里面的数据
pom依赖:
mysql mysql-connector-java 5.1.49 com.alibaba.ververica flink-connector-mysql-cdc 1.2.0
连接MySQL获取数据:
DebeziumSourceFunctionStartupOptions:sourceFunction = MySQLSource. builder() .hostname("hadoop101") .port(3306) .username("root") .password("root12345") // 可以指定多个库,用逗号隔开 .databaseList("gmall2021_realtime") // 巨坑,多个库时必须带上库名.表名,不然采集不到数据 .tableList("gmall2021_realtime.table_process") .serverTimeZone("Asia/Shanghai") .startupOptions(StartupOptions.latest( )) // 自定义反序列化器, 反序列化出自己想要的数据格式 .deserializer(new MyDebeziumDeserializerFunc( )) .build( );
initial:
第一次启动时 读取原表已有的历史数据, 操作类型为READ, 之后不断做检查点存储
第二次启动时 一定要指明检查点文件的具体位置, 这样就可以断点续传; 即使Flink宕机了, 重启后是从上次offset开始读, 而不是latest
检查点在打包部署后才有用, 因为那样才可以指明检查点的具体位置
earliest:
从BinLog第一行数据开始读, 最好先给这个数据库加上BinLog后, 再去读取创建数据库
latest:
读取最新变更数据, 从Flink程序启动后开始算
timestamp:
可以从BinLog某一时刻的数据开始读
specificOffset:
指明BinLog文件位置和从哪个offset开始读;
这个一般来说不怎么用, 因为本地没存offset的信息, 很难知道offset读到哪了
Flink-CDC中插入数据 的操作类型叫create
测试时:
用latest( ), 在Flink运行时再去操控数据库影响BinLog或者用initial也行, 不过Flink重启时没法指明检查点位置, 每次都得读原表,这部分操作类型都是READ
打包部署时: 因为Flink重启时可以从指明的检查点位置进行恢复, 故用initial( )
1.x VS 2.x: 1.x痛点:1.x在initial读取原表已有的历史数据时(全量读取阶段):
会对这部分数据加锁(不让别的线程来修改)来确保一致性
仅支持单线程读取
不支持CheckPoint
这要是中途失败了, 得从头开始读
2.x在initial读取原表已有的历史数据时(全量读取阶段):
- 不会对这部分数据加锁多线程读取支持CheckPoint
对于有主键的表, 全量读取阶段的数据 切分成一个个的Chunk, 每个Chunk单独做CheckPoint
各个Chunk可以并行读取
读取前记录BinLog的位置为低位点, 读取完毕后记录BinLog的位置为高位点;
用低位点到高位点之间的数据(即增量数据) 对刚才全量读取阶段读取到的数据进行修正
这样保证了最终一致性, 修正前的短期内 可能出现数据不一致
