TiCDC

我近半年的时间都在做 TiCDC Sink 模块的改造工作，目前新的 Sink 实现已经成功替换了旧的实现。最近有客户希望通过自己实现 Sink 的方式来接入 TiCDC，所以我想把这段时间的改造和设计经验分享出来，希望能帮助到大家。 此博客在 GitHub 上公开发布。 如果您有任何问题，请在此处打开一个 issue。 ⚠️ 注意： 该指南主要面向开发者，如果您只是想使用 TiCDC，请参阅 TiCDC 使用文档。 在阅读该指南前，请先阅读 TiCDC 架构和数据同步链路解析了解 TiCDC 的基本架构和数据同步流程。 基本概念 可以先简单浏览这些子组件概念，后面会有详细的介绍。 Sink：TiCDC 的 Sink 模块负责将 TiCDC 的数据变更输出到外部系统中。目前 TiCDC 支持输出到 MySQL、TiDB、Kafka、S3 等外部系统中。 Table Sink：负责将 TiCDC 的数据变更按照表为单位进行聚合，然后输出到外部系统中。 Event Sink：负责与外部系统进行交互，将 TiCDC 的数据变更编码后输出到外部系统中。这里的 Event 主要指的是 TiCDC 的数据变更事件，比如 Insert、Update、Delete 等。 MQ Event Sink：负责将 TiCDC 的数据变更输出到 Message Queue 中。MQ Sink 会将数据变更事件编码为 MQ 消息，然后输出到 MQ 中。目前 TiCDC 支持输出到 Kafka 中。 Txn Event Sink：负责将 TiCDC 的数据变更按照事务为单位进行聚合，然后输出到外部系统中。目前 TiCDC 支持输出到 MySQL、TiDB 中。 DDL Sink：负责将 TiCDC 的 DDL 语句输出到外部系统中。目前 TiCDC 支持输出到 Kafka、MySQL、TiDB 中。 MQ DDL Sink：负责将 TiCDC 接受到的 DDL 语句输出到 Kafka 中。 Txn DDL Sink：，负责将 TiCDC 接受到的 DDL 语句输出到 MySQL、TiDB 中。 基本架构 我们可以将 TiCDC 接收到的数据分为两类：...

TiCDC 是 TiDB 生态中的一个数据同步工具，它通过拉取 TiKV 的变更日志实现数据的增量同步。它除了可以将 TiDB 的数据同步至 MySQL 兼容的数据库之外，还提供了同步至 Kafka 的能力，支持 canal 和 avro 多种开放消息协议供其他系统订阅数据变更。 基本概念 以下这些都是理解 TiCDC 架构之前需要熟悉的基本概念，主要是对 TiDB 内部的一些组件和概念的解释。 TiDB: 下文中提到的 TiDB 主要是指它作为是一个计算层的抽象提供执行 SQL 的能力，没有真实的存储数据。 TiKV: 一个支持 ACID 的键值数据库，它可以作为 TiDB 的存储层。 Region: 它是 TiKV 数据移动的基本单位，可以将它认为是一组键值对集合。每个 Region 被复制到多个 TiKV 节点。 两阶段提交：一种分布式算法，它协调所有参与分布式原子事务的进程，决定是否提交或中止（回滚）该事务。 StartTs: TiDB 在开始两阶段提交时获取到的一个全局唯一递增的时间戳作为当前事务的唯一事务 ID，这里称为该事务的 start_ts。 CommitTs: TiDB 在提交两阶段提交事务时获取到另外一个全局唯一递增的时间戳作为该事务的 commit_ts。 什么是 CDC？ CDC 的全称为 Change Data Capture，它是指从源数据库捕获数据并且将其同步到其他数据库或者应用程序的过程。它作为一种很常见的数据集成方式被大量的应用在数据仓库中。当然任何的数据库系统都可以构建自己的 CDC，比如 SQL Server 的 CDC。TiCDC 就是专属于 TiDB 的 CDC，它的上游只能是 TiDB，但是它的下游可以是其他 MySQL 兼容的数据库系统，也可以是消息队列。...

最近在改造 TiCDC 的 Sink 组件时需要为 MQ 类型的 Sink 接上一个 unbound 的 channel。 在搜索过程中发现了一个项目叫做 chann，它是一个接口统一并且支持范型的 channel。 虽然这个库看上去实现很简单，但是我在实际使用过程中并不是很顺利。下面我就介绍一下我在使用该库时遇到的问题和进行的思考。 此博客在 GitHub 上公开发布。 如果您有任何问题或疑问，请在此处打开一个 issue。 简介 该库由 golang 社区非常活跃的 changkun 编写，托管在他组建的 golang-design 组织下。 它提供了统一的接口来创建不同类型的 channel，并且支持范型： ch := chann.New[int]() // 无界限, 无容量限制 ch := chann.New[func()](chann.Cap(0)) // 没有缓存, 容量为 0 ch := chann.New[string](chann.Cap(100)) // 有缓存，容量为 100 它的发送和接收操作和原生 channel 一致： ch.In() <- 42 println(<-ch.Out()) // 42 它的关闭接口为： ch.Close() 从接口来看，它几乎能无缝的接入到我当前的项目当中，这也是我选择它的原因。 源码阅读 在开始分析我遇到的问题之前，我们需要先深入阅读一下源码。它的源码不是很长，并且我主要是用的是无界限的 chann，所以下面就着重看一下无界限的 chann 的源码。 type Chann[T any] struct { in, out chan T close chan struct{} cfg *config q []T } Chann 是一个范型结构体，它里面维护了 in 和 out channel，这就是我们能使用原生 channel 语法操作 chann 的入口和出口。...

免责声明：我既不懂 TiCDC，也不懂 Flink。所以下面的文章很有可能是一派胡言。 最近我在帮 TiCDC 的 Kafka Sink 做多 Topic 支持，目标是让 TiCDC 能够把 TiDB 的数据接入到 Flink。其实现在的 TiCDC 就具备接入 Flink 的能力，只是受限于目前的 Kafka Sink 只能支持单个 Topic，接入和维护非常麻烦。 我来介绍一下目前 TiCDC 接入 Flink 的现状和方法。 此博客在 GitHub 上公开发布. 如果您有任何问题或疑问，请在此处打开一个 issue. 简介 TiCDC Kafka Sink 支持多种协议格式，其中活跃维护的是 TiCDC 自己实现的 open-protocol 和阿里巴巴的 canal-json 协议。这次我们需要使用 canal-json 协议并通过 Flink 的 Kafka connector 接入 Flink 系统。 启动 TiDB 和 TiCDC 集群 推荐使用 TiUP 启动 TiDB 和 TiCDC 集群，因为今天只是简单演示所以我就用 TiUP 的 Playground 启动测试集群。...