关于数据同步:解读重要功能特性新手入门-Apache-SeaTunnel-CDC

点亮 ⭐️ Star · 照亮开源之路

https://github.com/apache/inc…

为什么说 CDC 是 SeaTunnel 平台中的一个重要性能个性？明天这篇文章跟大家分享一下 CDC 是什么？目前市面上的 CDC 工具现有的痛点有哪些？SeaTunnel 面对这些痛点设计的架构指标是什么？另外包含社区的瞻望和目前在做的一些事件。

总体来说，市面上曾经有这么多 CDC 工具了，咱们为什么还要反复去造一个轮子？

带着这个疑难，我先给大家简要介绍下 CDC 是什么！CDC 的全称是 Change Data Capture，它就是一个数据变更捕捉。变更数据捕捉 (CDC) 应用 Server 代理来记录利用于表的插入、更新和删除流动。这样，就能够按易于应用的关系格局提供这些更改的详细信息。将为批改的行捕捉列信息以及将更改利用于指标环境所需的元数据，并将其存储在镜像所跟踪源表的列构造的更改表中。

在 MySQL，PostgreSQL，MongoDB 等等数据库之间相互同步数据，或者把这些数据库的数据同步到 Elasticsearch 里以供全文搜寻，当然也能够基于 CDC 对数据库进行备份。而数据分析系统能够通过订阅感兴趣的数据表的变更，来获取所须要的剖析数据进行解决，不须要把剖析流程嵌入到已有零碎中，以实现解耦。

在微服务大行其道的今日，微服务之间信息共享始终比较复杂，CDC 也是一种可能的解决方案，微服务能够通过 CDC 来获取其余微服务数据库的变更，从而获取数据的状态更新，执行本人相应的逻辑。

通常缓存更新都比拟难搞，能够通过 CDC 来获取数据库的数据更新事件，从而管制对缓存的刷新或生效。

而 CQRS 是什么又是一个很大的话题，简略来讲，你能够把 CQRS 了解为一种高配版的读写拆散的设计模式。举个例子，咱们后面讲了能够利用 CDC 将 MySQL 的数据同步到 Elasticsearch 中以供搜寻，在这样的架构里，所有的查问都用 ES 来查，但在想批改数据时，并不间接批改 ES 里的数据，而是批改上游的 MySQL 数据，使之产生数据更新事件，事件被消费者生产来更新 ES 中的数据，这就基本上是一种 CQRS 模式。而在其余 CQRS 的零碎中，也能够利用相似的形式来更新查问视图。

开源组件	Canal	Debezium	Flink CDC
反对数据库	仅反对 MySQL	反对 MySQL、Postgre SQL、Oracle 等	反对 MySQL、Postgre SQL、Oracle 等
同步历史数据	不反对	单并行锁表	多并行无锁
输入端	Kafka、RocketMQ	Kafka	Flink Connector

Canal

数据库它仅反对 MySQL，不反对同步历史数据，只能同步增量数据，输入端除了反对 canal client/adapter（适配工作量很大），还反对了的 Kafka 和 RocketMQ。

Debezium

反对的数据库比拟多，不仅反对 MySQL，PG，Oracle，还反对其它 Mongo DB 等数据库，同时反对同步历史数据，不过历史数据读取形式是：一个快照读整个表，如果你表很大，就会像 sqoop 一样读特地久。如果中途失败了，须要从头开始读，这样会呈现一些问题。而且输入端上反对的就更加少，仅仅反对通过 Kafka 输入。

Flink CDC

Flink CDC 和前两个定位上就不一样。它理论就是 Flink 生态的 connector，就是连接器组。目前也反对比拟多的数据库，像 MySQL PG，Oracle，Mongo 这些数据库都是反对的。

绝对于后面的开源组件，它持一个多边形无锁的算法。当然它也是参考到 Netflix DBLog 的无锁算法。因为它是基于 Flink 生态的，所以它输入端就比拟多。只有是 Flink 生态有的 connector，反对 Upsert 的 Connector 都是能够应用的。当然它也会存在很多问题，这个问题就是前面我会提到的。

如果用过 Flink CDC 的敌人就会发现，咱们须要对每一个表进行配置。比方咱们想同步 10 张表，就要写 10 个 source 的 SQL，10 个 sink 的 SQL，如果你要进行 transform，就还要写 transform 的 SQL。

这个状况下，小数量的表手写还能够应酬，如果数量大可能就呈现类型映射谬误的问题，或者参数配置谬误的问题，就会产生很高的运维老本（配置麻烦）。而 Apache SeaTunnel 定位就是一个简略易用的数据集成平台，咱们冀望解决这个问题。

支不反对 schema 的变更。实际上像 Flink CDC 和 Debezium，两者反对 DDL 事件发送，然而不反对发送到 Sink，让 Sink 做同步变更。或者 Fink CDC 能拿到事件，然而无奈发送到引擎中，因为引擎不能基于 DDL 事件去变更 transform 的 Type information，Sink 没方法跟着 DDL 事件进行变更。

如果有 100 张表，因为 Flink CDC 只反对一个 source 去同步一张表，每一张表都会应用一个链接，当表多的时候，应用的链接就特地多，就会对源头的 JDBC 数据库造成了很大的连贯压力，并且会持有特地多的 Binlog，也会像 worker 这种，也还会造成反复的日志解析。

SeaTunnel CDC 是基于市面上现有的 CDC 组件的优缺点，以及相干痛点问题做的架构设计。

反对根底的 CDC
反对无锁并行快照历史数据
反对日志心跳检测和动静加表
反对分库分表和多构造表读取
反对 Schema evolution

反对增量日志的读取，还至多要可能反对无锁并行快照历史数据的能力。

咱们冀望可能缩小用户的运维老本，可能动静的加表，比方有时候想同步整个库，前面新增了一张表，你不须要手动去保护，能够不必再去改 Job 配置，也不必进行 Job 再重启一遍，这样就会缩小很多麻烦。

反对分库分表和多构造表的读取，其实这也是咱们最开始提到的每个表独自配置的问题。并且还反对 Schema evolution，DDL 的传输，还有在引擎中能反对 schema evolution 的变更，可能变更到 Transform 和 Sink 下面去。

CDC 根底流程蕴含:

快照阶段：用于读取表的历史数据
- 最小 Split 粒度: 表的主键范畴数据
增量阶段：用于读取表的增量日志更改数据
- 最小 Split 粒度: 以表为单位

枚举器生成一个表的多个 SnapshotSplit，并将它们调配给 reader。

//  pseudo-code. 
public class SnapshotSplit implements SourceSplit {
    private final String splitId;
    private final TableId tableId;
    private final SeaTunnelRowType splitKeyType;
    private final Object splitStart;
    private final Object splitEnd;
}

当 SnapshotSplit 读取实现时，读取器将拆分的高水位线报告给枚举器。当所有 SnapshotSplit 都报告高水位线时，枚举器开始增量阶段。

//  pseudo-code. 
public class CompletedSnapshotSplitReportEvent implements SourceEvent {
    private final String splitId;
    private final Offset highWatermark;
}

有 4 个步骤：

日志低水位线：读取快照数据前获取以后日志偏移量。
读取 SnapshotSplit 数据：读取属于 split 的数据范畴，这里分为两种状况

- 案例 1：步骤 1 &2 不能原子化（MySQL）
  
  因为咱们不能加表锁，也不能加基于低水位线的区间锁，所以第 1 步和第 2 步不是孤立的。
- exactly-once：应用内存表保留历史数据 & 过滤日志数据从低水位线到高水位线
- At-least-once：间接输入数据并应用低水位线而不是高水位线
- 案例 2：步骤 1 和 2 能够原子化（Oracle）
  
  能够应用 for scn 来保障两步的原子化
- Exactly-Once：间接输入数据并应用低水位线而不必去获取高水位线

加载高水位线数据：
- 步骤 2 中案例 1 & Exactly-Once：读取快照数据后获取以后日志偏移量。
- 其余：应用低水位线代替高水位线
如果高水位线 > 低水位线，读取范畴日志数据

因为咱们无奈确定查问语句在高下水位之间执行的地位，为了保证数据的 exactly-once，咱们须要应用内存表来长期保留数据。

日志低水位线：读取快照数据前获取以后日志偏移量。
读取 SnapshotSplit 数据：读取属于 split 的范畴数据，写入内存表。
日志高水位线：读取快照数据后获取以后日志偏移量。
读取范畴日志数据：读取日志数据并写入内存表
输入内存表的数据，开释内存使用量。

当所有快照拆分报告水位时，开始增量阶段。

联合所有快照拆分和水位信息，取得 LogSplits。

咱们心愿最小化日志连贯的数量：

增量阶段默认只有一个 reader 工作，用户也能够依据需要去配置选项指定数量（不能超过 reader 数量）
一个 reader 最多取得一个连贯

//  pseudo-code. 
public class LogSplit implements SourceSplit {
    private final String splitId;
    /**
     * All the tables that this log split needs to capture.
     */
    private final List<TableId> tableIds;
    /**
     * Minimum watermark for SnapshotSplits for all tables in this LogSplit
     */
    private final Offset startingOffset;
    /**
     * Obtained by configuration, may not end
     */
    private final Offset endingOffset;
    /**
     * SnapshotSplit information for all tables in this LogSplit.
     * </br> Used to support Exactly-Once.
     */
    private final List<CompletedSnapshotSplitInfo> completedSnapshotSplitInfos;
    /**
     * Maximum watermark in SnapshotSplits per table.
     * </br> Used to delete information in completedSnapshotSplitInfos, reducing state size.
     * </br> Used to support Exactly-Once.
     */
    private final Map<TableId, Offset> tableWatermarks;
}

//  pseudo-code. 
public class CompletedSnapshotSplitInfo implements Serializable {
    private final String splitId;
    private final TableId tableId;
    private final SeaTunnelRowType splitKeyType;
    private final Object splitStart;
    private final Object splitEnd;
    private final Offset watermark;
}

Exactly-Once：

阶段 1：在水印数据之前应用 completedSnapshotSplitInfos 过滤器。
阶段 2：表不再须要过滤，在 completedSnapshotSplitInfos 中删除属于该表的数据，因为前面的数据须要解决。

At-Least-Once：无需过滤数据，且 completedSnapshotSplitInfos 不须要任何数据

场景 1：发现新表时，枚举器处于快照阶段，间接调配新的 split。

场景 2：发现新表时，枚举器处于增量阶段。

在增量阶段动静发现新表。

暂停 LogSplit reader。
Reader 暂停运行。
Reader 报告以后日志偏移量。
将 SnapshotSplit 调配给阅读器。
Reader 执行快照阶段读取。
Reader 报告所有 SnapshotSplit 水位。
为 Reader 调配一个新的 LogSplit。
Reader 再次开始增量读取并向枚举器确认。

多构造表是为了解决连接器实例过多，配置过于简单的问题。比方你只须要去配表的一个正则，或者配多个表名，不须要对每一个表去做配置。

长处：占用数据库连贯少，缩小数据库压力
毛病：在 SeaTunnel Engine 中，多个表会在一个管道中，容错的粒度会变大。

这个个性容许 Source 反对读取多个构造表，再应用侧流输入与单表流保持一致。Sink 如果也去反对多表，可能波及改变比拟多。所以第一阶段的指标只是让 Source 去反对多构造表，这里配置的逻辑可能会和原来的不一样，会通过 catalog 去读每一个 config 外面到底配了哪些表，再把表塞到 Source Connector 中，这里多表构造的 API 曾经在 SeaTunnel 的 API 之中，然而还没有做相干的适配。

目前开发实现的是 CDC 的根底能力，可能反对增量阶段和快照阶段，MySQL 也曾经反对了，反对实时和离线。MySQL 实时曾经测试实现了，离线的测试还没有实现。Schema 因为要波及到 Transfrom 和 Sink 的变更，目前还没有反对的。动静发现新表还没有反对，多构造表目前曾经预留了一些接口进去，然而适配的工作量比拟大，可能等到 2023 年 Q1 季度可能会做这个事件。

作为一个 Apache 孵化我的项目，Apache SeaTunnel 社区迅速倒退，在接下来的社区规划中，次要有四个方向：

扩充与欠缺 Connector & Catalog 生态

反对更多 Connector & Catalog，如 TiDB、Doris、Stripe 等，并欠缺现有的连接器，进步其可用性与性能等；

反对 CDC 连接器，用于反对实时增量同步场景；

对连接器感兴趣的同学能够关注该 Umbrella：https://github.com/apache/inc…
反对引擎的更多版本

如 Spark 3.x, Flink 1.14.x 等

对反对 Spark 3.3 感兴趣的同学能够关注该 PR：https://github.com/apache/inc…
反对更多数据集成场景（SeaTunnel Engine）

用于解决整库同步、表构造变更同步、工作失败影响粒度大等现有引擎不能解决的痛点；

对 engine 感兴趣的同学能够关注该 Umbrella：https://github.com/apache/inc…
更简略易用（SeaTunnel Web）

提供 Web 界面以 DAG/SQL 等形式使操作更简略，更加直观的展现 Catalog、Connector、Job 等；

接入调度平台，使工作治理更简略；

对 Web 感兴趣的同学能够关注咱们的 Web 子项目：https://github.com/apache/inc…

Apache SeaTunnel(Incubating) 是一个分布式、高性能、易扩大、用于海量数据（离线 & 实时）同步和转化的数据集成平台

仓库地址： https://github.com/apache/inc…

网址：https://seatunnel.apache.org/

Proposal：https://cwiki.apache.org/conf…

Apache SeaTunnel (Incubating) 下载地址 ：https://seatunnel.apache.org/…

衷心欢送更多人退出！

咱们置信，在 「Community Over Code」（社区大于代码）、「Open and Cooperation」（凋谢合作）、「Meritocracy」（精英治理）、以及「 多样性与共识决策」 等 The Apache Way 的指引下，咱们将迎来更加多元化和容纳的社区生态，共建开源精力带来的技术提高！

咱们诚邀各位有志于让外乡开源立足寰球的搭档退出 SeaTunnel 贡献者小家庭，一起共建开源！

提交问题和倡议：https://github.com/apache/inc…

奉献代码：https://github.com/apache/inc…

订阅社区开发邮件列表 : dev-subscribe@seatunnel.apach…

开发邮件列表：dev@seatunnel.apache.org

退出 Slack:https://join.slack.com/t/apac…

关注 Twitter: https://twitter.com/ASFSeaTunnel

关于数据同步:解读重要功能特性新手入门-Apache-SeaTunnel-CDC

引言

CDC 的应用场景

异构数据库之间的数据同步或备份 / 建设数据分析计算平台

微服务之间共享数据状态

更新缓存 / CQRS 的 Query 视图更新

现有 CDC 组件

现有组件存在的痛点

单表配置

不反对 Schema Evolution

持有链接过多

SeaTunnel CDC 架构指标

CDC 根本流程

快照阶段

快照阶段 – SnapshotSplit 读取流程

快照阶段—MySQL Snapshot Read & Exactly-once

增量阶段

动静发现新表

多构造表同步

SeaTunnel CDC 现状

Apache SeaTunnel 瞻望

Apache SeaTunnel