Flink CDC 2.0 主要是借鉴 DBLog 算法-CSDN博客

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

DBLog 算法原理

DBLog 这个算法的原理分成两个部分第一部分是分 chunk第二部分是读 chunk。分 chunk 就是把一张表分为多个 chunk桶/片。我可以把这些 chunk 分发给不同的并发的 task 去做。例如有 reader1 和 reader2不同的 reader 负责读不同的 chunk。其实只要保证每个 reader 读的那个 chunk 是完整的也能跟最新的 Binlog 能够匹配在一起就可以了。在读 chunk 的过程中会同时读属于这个 chunk的历史数据也会读这个 chunk 期间发生的 Binlog 事件然后来做一个 normalize。

首先是 chunk 的划分。一张表它的 ID 字段是主键 PK。通过 query 能够知道最大的 PK 也能知道最小的 PK。然后根据最大最小的 PK 设定一个步长那么就能把这个表分成一些 chunk。每个 chunk 是一个左闭右开的区间这样可以实现 chunk 的无缝衔接。第一个 chunk 和最后一个 chunk 最后一个字段是一个正无穷和负无穷的表示即所有小于 k1 的 key 由第一个 chunk 去读所有大于 K104 的 key 由最后一个chunk去读。

chunk 的读取首先有一个 open 打点的过程还有一个 close 打点的过程。例如读属于这个 chunk1 的所有数据时橘色的 K1 到 K7 是这些全量数据。橘黄色里面有下划线的数据是在读期间这些 Binlog 在做改变。比如 K2 就是一条 update从 100 变成 108K3 是一条 delete。K2 后面又变成 119。还有 K5 也是一个update。在 K2、K3、K5 做标记说明它们已经不是最新的数据了需要从 Binlog 里面读出来做一个 merge 获取最新的数据最后读出来的就是在 close 位点时刻的最新数据。最后的效果就是将 update 最新的数据最终输出将 delete 的数据如 K3 不输出。所以在 chunk1 读完的时候输出的数据是 K0、K2、K4、K5、K6、K7 这些数据这些数据也是在 close 位点时数据库中该 chunk 的数据。

接下来是 chunk 的分发。一张表切成了 N 个 chunk 后SourceEnumerator 会将这些 chunk 分给一些 SourceReader 并行地读取并行读是用户可以配置的这就是水平扩展能力。

每一个 Snapshot chunk 读完了之后有一个信息汇报过程这个汇报非常关键包含该 chunk 的基本信息和该 chunk 是在什么位点读完的即 close 位点。在进入 Binlog 读取阶段之后在 close 位点之后且属于这个 chunk 的 binlog 数据是要需要继续读取的从而来保证数据的完整性。

在所有的 Snapshot chunk 读完之后会发一个特殊的 Binlog chunk该 chunk 里包含刚刚所有 Snapshot chunk 的汇报信息。Binlog Reader 会根据所有的 Snapshot chunk 汇报信息按照各自的位点进行跳读跳读完后再进入一个纯粹的 binlog 读取。跳读就是需要考虑各个 snapshot chunk 读完全量时的 close 位点进行过滤避免重复数据纯 binlog 读就是在跳读完成后只要是属于目标表的 changelog 都读取。

Flink CDC 增量快照算法流程为首先一张表按 key 分成一个个 chunkBinlog在不断地写全量阶段由 SourceReader 去读进入增量阶段后SourceReader 中会启一个 BinlogReader 来读增量的部分。全量阶段只会有 insert only 的数据增量阶段才会有 update、delete 数据。SourceReader 中具体去负责读 chunk 的 reader 会根据收到的分片类型决定启动 SnapshotReader 还是 BinlogReader。

Flink CDC 增量快照算法的核心价值包括

第一实现了并行读取水平扩展的能力即使十亿百亿的大表只要资源够那么水平扩展就能够提升效率

第二实现 Dblog 算法的变动它能够做到在保证一致性的情况下实现无锁切换

第三基于 Flink 的 state 和 checkpoint 机制它实现了断点续传。比如 task 1 失败了不影响其它正在读的task只需要把 task 1 负责的那几个 chunk 进行重读

第四全增量一体化全增量自动切换。

增量快照框架的设计围绕一个核心的 API 展开这个核心的 API 就是DataSourceDialect数据源方言这个方言关心的就是面向某个数据源特有的在接入全增量框架时需要实现的方法。

比较常见的CDC工具大都有过使用经验

Debezium是国外⽤户常⽤的CDC组件单机对于分布式来说在数据读取能力的拓展上没有分布式的更具有优势在大数据众多的分布式框架中Hive、Hudi等Flink CDC 的架构能够很好地接入这些框架。

DataX无法支持增量同步。如果一张Mysql表每天增量的数据是不同天的数据并且没有办法确定它的产生时间那么如何将数据同步到数仓是一个值得考虑的问题。DataX支持全表同步也支持sql查询的方式导入导出全量同步一定是不可取的sql查询的方式没有可以确定增量数据的字段的话也不是一个好的增量数据同步方案。

Canal是用java开发的基于数据库增量日志解析提供增量数据订阅&消费的中间件。Canal主要支持了MySQL的Binlog解析将增量数据写入中间件中例如kafka,Rocket MQ等但是无法同步历史数据因为无法获取到binlog的变更。

Sqoop主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql...)间进行数据的传递。Sqoop将导入或导出命令翻译成mapreduce程序来实现这样的弊端就是Sqoop只能做批量导入遵循事务的一致性Mapreduce任务成功则同步成功失败则全部同步失败。

Apache SeaTunnel是一个当前也非常受欢迎的数据集成同步组件。其可以支持全量和增量支持流批一体。SeaTunnel的使用是非常简单的零编写代码只需要写一个配置文件脚本提交命令即可同时也使用分布式的架构可以依托于Flink,Spark以及自身的Zeta引擎的分布式完成一个任务在多个节点上运行。其内部也有类似Flink checkpoint的状态保存机制用于故障恢复sink阶段的两阶段提交机制也可以做到精准一次性Excatly-once。对于大部分的场景SeaTunnel都能完美支持但是SeaTunnel只能支持简单的数据转换逻辑对于复杂的数据转换场景还是需要Flink、Spark任务来完成。

Flink CDC 基本都弥补了以上框架的不足将数据库的全量和增量数据一体化地同步到消息队列和数据仓库中也可以用于实时数据集成将数据库数据实时入湖入仓无需像其他的CDC工具一样需要在服务器上进行部署减少了维护成本链路更少完美套接Flink程序CDC获取到的数据流直接对接Flink进行数据加工处理一套代码即可完成对数据的抽取转换和写出既可以使用flink的DataStream API完成编码也可以使用较为上层的FlinkSQL API进行操作。

DBlog paper 论文的 chunk 切分算法。

该算法在数据库中维护了一张watermark信号表记录每个chunk块的区间值位点LW和HW。

Flink CDC2.x 并没有维护信号表通过直接读取 binlog 位点替代在 binlog 中做标记的功能。

左边是 Chunk 的切分算法描述Chunk 的切分算法其实和很多数据库的分库分表原理类似通过表的主键对表中的数据进行分片。假设每个 Chunk 的步长为 10按照这个规则进行切分只需要把这些 Chunk 的区间做成左开右闭或者左闭右开的区间保证衔接后的区间能够等于表的主键区间即可。

右边是每个 Chunk 的无锁读算法描述该算法的核心思想是在划分了 Chunk 后对于每个 Chunk 的全量读取和增量读取在不用锁的条件下完成一致性的合并。

因为每个 chunk 只负责自己主键范围内的数据不难推导只要能够保证每个 Chunk 读取的一致性就能保证整张表读取的一致性这便是无锁算法的基本原理。

Netflix 的 DBLog 论文中 Chunk 读取算法是通过在 DB 维护一张信号表再通过信号表在 binlog 文件中打点记录每个 chunk 读取前的 Low Position (低位点) 和读取结束之后 High Position (高位点) 在低位点和高位点之间去查询该 Chunk 的全量数据。在读取出这一部分 Chunk 的数据之后再将这 2 个位点之间的 binlog 增量数据合并到 chunk 所属的全量数据从而得到高位点时刻该 chunk 对应的全量数据。

Flink CDC 结合自身的情况在 Chunk 读取算法上做了去信号表的改进不需要额外维护信号表通过直接读取 binlog 位点替代在 binlog 中做标记的功能整体的 chunk 读算法描述如下图所示

比如正在读取 Chunk-1Chunk 的区间是 [K1, K10]首先直接将该区间内的数据 select 出来并把它存在 buffer 中在 select 之前记录 binlog 的一个位点 (低位点)select 完成后记录 binlog 的一个位点 (高位点)。然后开始增量部分消费从低位点到高位点的 binlog。

图中的 - ( k2,100 ) + ( k2,108 ) 记录表示这条数据的值从 100 更新到 108
第二条记录是删除 k3
第三条记录是更新 k2 为 119
第四条记录是 k5 的数据由原来的 77 变更为 100。

观察图片中右下角最终的输出会发现在消费该 chunk 的 binlog 时出现的 key 是k2、k3、k5我们前往 buffer 将这些 key 做标记。

对于 k1、k4、k6、k7 来说在高位点读取完毕之后这些记录没有变化过所以这些数据是可以直接输出的
对于改变过的数据则需要将增量的数据合并到全量的数据中只保留合并后的最终数据。例如k2 最终的结果是 119 那么只需要输出 +(k2,119)而不需要中间发生过改变的数据。

通过这种方式Chunk 最终的输出就是在高位点是 chunk 中最新的数据。

上图描述的是单个 Chunk 的一致性读但是如果有多个表分了很多不同的 Chunk且这些 Chunk 分发到了不同的 task 中那么如何分发 Chunk 并保证全局一致性读呢

有 SourceEnumerator 的组件这个组件主要用于 Chunk 的划分划分好的 Chunk 会提供给下游的 SourceReader 去读取通过把 chunk 分发给不同的 SourceReader 便实现了并发读取 Snapshot Chunk 的过程同时基于 FLIP-27 我们能较为方便地做到 chunk 粒度的 checkpoint。

当 Snapshot Chunk 读取完成之后需要有一个汇报的流程如下图中橘色的汇报信息将 Snapshot Chunk 完成信息汇报给 SourceEnumerator。

汇报的主要目的是为了后续分发 binlog chunk (如下图)。因为 Flink CDC 支持全量 + 增量同步所以当所有 Snapshot Chunk 读取完成之后还需要消费增量的 binlog这是通过下发一个 binlog chunk 给任意一个 Source Reader 进行单并发读取实现的。

整体流程可以概括为首先通过主键对表进行 Snapshot Chunk 划分再将 Snapshot Chunk 分发给多个 SourceReader每个 Snapshot Chunk 读取时通过算法实现无锁条件下的一致性读SourceReader 读取时支持 chunk 粒度的 checkpoint在所有 Snapshot Chunk 读取完成后下发一个 binlog chunk 进行增量部分的 binlog 读取这便是 Flink CDC 2.0 的整体流程如下图所示

MySQL CDC 2.0核心feature 包括

并发读取全量数据的读取性能可以水平扩展
全程无锁不对线上业务产生锁的风险
断点续传支持全量阶段的 checkpoint。

SQL server CDC

CDC实现的原理

① 数据源监控的源表所有操作都会进入日志。日志为变更表提供数据源作为捕获进程的输入来源。

② 捕获进程扫描日志并将列数据以及与事务有关的信息写入变更数据捕获更改表中。

③ 捕获进程将在每个扫描周期内打开并提交其自己的事务。它检测何时为表新启用了变更数据捕获并自动将这些表加入到当前在日志中监视更改项的表集中。同样它还会检测禁用的变更数据捕获进而从当前监视更改数据的表集中删除源表。在处理完日志的某个部分后捕获进程将通知服务器日志截断逻辑后者使用此信息来确定适合截断的日志项所以采用完整备份和简单备份日志方式对CDC捕获不会产生影响。

SQL Server CDC 长什么样

原始日志

常见的数据库往往存在以下两种日志

redo 日志
- 记录数据的正向变更简单来说事务的 commit 通常先记录在这个文件再返回应用程序成功可确保数据 持久性
undo 日志
- 用于保证事务的 原子性如执行 rollback 命令即反向执行 undo 日志中内容以达成数据回滚

一条 DML 语句写入数据库流程如下

大部分关系型数据库中一个或多个变更会被隐式或显式包装成一个事务
事务开始数据库引擎定位到数据行所在的 文件位置 并根据已有的数据生成 前镜像 和 后镜像
后镜像 数据记录到 redo 日志中前镜像 数据记录到 undo 日志中
事务提交后日志提交位点检查点向前推进已提交的日志内容即可能被覆盖或者释放

SQL Server redo/undo 日志采用了 ldf 格式 ,文件循环使用。

ldf 日志文件由多个 VLF(逻辑日志) 组合在一起这些 VLF 首尾相连形成完整的数据库日志记录
ldf 在逻辑日志末端到达物理日志文件末端时新的日志记录将回到物理日志文件开始复写旧的数据

ldf 文件即 CDC 所分析的增量日志文件。

cdc.console_capture
- 负责分析 ldf 日志 并解析 console 数据库事件,再将其写入到 CDC 表中
- 间隔 5 秒钟执行一次扫描每次扫描 10 轮每轮扫描最多 500 个事务
cdc.console_cleanup
- 负责定期清理 CDC 表中较老的数据
- 默认保留 3 天 CDC 日志数据4320秒

开启 CDC 功能后SQL Server 数据库会多出一个名称为 cdc 的 schema里面会多出下列这些表。