解读重要功能特性：新手入门 Apache SeaTunnel CDC

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

https://github.com/apache/incubator-seatunnel

引言

CDC的使用场景

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

微服务之间共享数据状态

更新缓存 CQRS 的 Query 视图更新

通常缓存更新都比较难搞，可以通过 CDC 来获取数据库的数据更新事件，从而控制对缓存的刷新或失效。

而 CQRS 是什么又是一个很大的话题，简单来讲，你可以把 CQRS 理解为一种高配版的读写分离的设计模式。举个例子，我们前面讲了可以利用 CDC 将 MySQL 的数据同步到 Elasticsearch 中以供搜索，在这样的架构里，所有的查询都用 ES 来查，但在想修改数据时，并不直接修改 ES 里的数据，而是修改上游的 MySQL 数据，使之产生数据更新事件，事件被消费者消费来更新 ES 中的数据，这就基本上是一种 CQRS 模式。而在其他 CQRS 的系统中，也可以利用类似的方式来更新查询视图。

现有CDC组件

Canal

Debezium

Flink CDC

现有组件存在的痛点

 单表配置

不支持 Schema Evolution

持有链接过多

SeaTunnel CDC架构目标

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

• 支持基础的CDC
• 支持无锁并行快照历史数据
• 支持日志心跳检测和动态加表
• 支持分库分表和多结构表读取
• 支持Schema evolution

CDC 基本流程

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

快照阶段

//  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;
}

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

快照阶段 - SnapshotSplit 读取流程

1. 日志低水位线：读取快照数据前获取当前日志偏移量。
2. 读取 SnapshotSplit 数据：读取属于split 的数据范围，这里分为两种情况

• 案例1：步骤1&2不能原子化（MySQL）

  因为我们不能加表锁，也不能加基于低水位线的区间锁，所以第 1 步和第 2 步不是孤立的。

• exactly-once：使用内存表保存历史数据 & 过滤日志数据从低水位线到高水位线
• At-least-once：直接输出数据并使用低水位线而不是高水位线
• 案例 2：步骤 1 和 2 可以原子化（Oracle）

  可以使用 for scn 来保证两步的原子化

• Exactly-Once：直接输出数据并使用低水位线而不用去获取高水位线

1. 加载高水位线数据：
• 步骤 2 中案例 1 & Exactly-Once：读取快照数据后获取当前日志偏移量。
• 其他：使用低水位线代替高水位线
2. 如果高水位线>低水位线，读取范围日志数据

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

1. 日志低水位线：读取快照数据前获取当前日志偏移量。
2. 读取 SnapshotSplit 数据：读取属于 split 的范围数据，写入内存表。
3. 日志高水位线：读取快照数据后获取当前日志偏移量。
4. 读取范围日志数据：读取日志数据并写入内存表
5. 输出内存表的数据，释放内存使用量。

增量阶段

• 增量阶段默认只有一个 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;
}

• 阶段 1：在水印数据之前使用 completedSnapshotSplitInfos 过滤器。
• 阶段2：表不再需要过滤，在 completedSnapshotSplitInfos 中删除属于该表的数据，因为后面的数据需要处理。

动态发现新表

add-table

1. 暂停 LogSplit reader。
2. Reader 暂停运行。
3. Reader 报告当前日志偏移量。
4. 将 SnapshotSplit 分配给阅读器。
5. Reader 执行快照阶段读取。
6. Reader 报告所有 SnapshotSplit 水位。
7. 为 Reader 分配一个新的 LogSplit。
8. Reader 再次开始增量读取并向枚举器确认。

多结构表同步

• 优点：占用数据库连接少，减少数据库压力
• 缺点：在 SeaTunnel Engine 中，多个表会在一个管道中，容错的粒度会变大。

SeaTunnel CDC现状

目前开发完成的是 CDC 的基础能力，能够支持增量阶段和快照阶段， MySQL 也已经支持了，支持实时和离线。MySQL 实时已经测试完成了，离线的测试还没有完成。Schema 因为要涉及到Transfrom 和Sink 的变更，目前还没有支持的。动态发现新表还没有支持，多结构表目前已经预留了一些接口出来，但是适配的工作量比较大，可能等到 2023 年 Q1 季度可能会做这个事情。

Apache SeaTunnel 展望

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

• 扩大与完善 Connector & Catalog 生态

  支持更多 Connector & Catalog，如TiDB、Doris、Stripe等，并完善现有的连接器，提高其可用性与性能等；

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

  对连接器感兴趣的同学可以关注该Umbrella：https://github.com/apache/incubator-seatunnel/issues/1946

• 支持引擎的更多版本

  如Spark 3.x, Flink 1.14.x等

  对支持Spark 3.3 感兴趣的同学可以关注该PR：https://github.com/apache/incubator-seatunnel/pull/2574

• 支持更多数据集成场景 （SeaTunnel Engine）

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

  对engine感兴趣的同学可以关注该Umbrella：https://github.com/apache/incubator-seatunnel/issues/2272

• 更简单易用（SeaTunnel Web）

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

  接入调度平台，使任务管理更简单；

  对Web 感兴趣的同学可以关注我们的Web子项目：https://github.com/apache/incubator-seatunnel-web

动态线程共享，SeaTunnel-TaskExecutionServer 的设计思路

SeaTunnel 在天翼云数据集成平台的探索实践

从 0 到 1 快速入门 Apache SeaTunnel ，新一代数据集成平台的原理和实践

你这么可爱，点个赞吧！