TiDB TiCDC 的应用实践

TiCDC 是一个通过拉取 TiKV 日志实现的 TiDB 增量数据同步工具，具有还原数据到与上游任意 TSO 一致状态的能力，同时提供开放数据协议，支持其他系统订阅数据变更。TiCDC 运行时是无状态的，借助 PD 内部的 etcd 实现高可用。TiCDC 集群支持创建多个同步任务，向多个不同的下游进行数据同步。

主要优势：

数据高可用：TiCDC 从 TiKV 获取变更日志，意味着只要 TiKV 具备高可用就能保证数据的高可用，遇到 TiCDC 全部异常关闭的极端情况，后续启动还能正常获取数据。

水平扩展：支持组建多 TiCDC 节点集群，将同步任务均匀地调度到不同节点上，面对海量数据时，可以通过添加节点解决同步压力过大的问题。

自动故障转移：当集群中的一个 TiCDC 节点意外退出时，该节点上的同步任务会自动调度到其余的 TiCDC 节点上。

支持多种下游系统和输出多种格式：目前已经支持兼容 MySQL 协议的数据库、Kafka 和 Pulsar 分布式流处理系统，支持输出的格式有 Apache Avro，Maxwell 和 Canal。

应用场景：

两中心主备

数据库作为企业 IT 的核心，在稳定运行的基础之上，数据库的容灾建设成为保障业务连续性的前提条件。

综合考量业务关键性、成本等因素，部分用户希望核心数据库只需要完成主备站点的容灾即可，利用 TiCDC 构建 TiDB 主备站点的容灾方案成为理想之选。该方案基于 TiCDC 的数据同步功能，可以适配两个中心长距离间隔、网络延迟较大的场景，进行两个数据中心 TiDB 集群之间的数据单向同步，保障事务的最终一致性，实现秒级 RPO。

环形同步与多活

利用 TiCDC 实现三个 TiDB 集群之间的环形同步，构建 TiDB 多中心的容灾方案。当一个数据中心的机柜发生意外掉电，可以把业务切换到另一个数据中心的 TiDB 集群，实现事务的最终一致性和秒级 RPO。为了分担业务访问的压力，在应用层可以随时切换路由，将流量切换到目前负载较小的 TiDB 集群提供服务，实现负载均衡，在满足数据高可用的同时提升容灾能力。

数据订阅

TiCDC 为下游数据消费端提供实时、高吞吐、稳定的数据订阅服务，通过开放数据协议（Open Protocol ）与 MySQL、Kafka、Pulsar、Flink、Canal、Maxwell 等多种异构生态系统对接，满足用户在大数据场景中对各类数据的应用与分析需求，广泛适用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等场景。

系统角色

TiKV CDC 组件：只输出 key-value (KV) change log。

内部逻辑拼装 KV change log。提供输出 KV change log 的接口，发送数据包括实时 change log 和增量扫的 change log。

capture：TiCDC 运行进程，多个 capture 组成一个 TiCDC 集群，负责 KV change log 的同步。

每个 capture 负责拉取一部分 KV change log。

对拉取的一个或多个 KV change log 进行排序。

向下游还原事务或按照 TiCDC Open Protocol 进行输出。

1部署TiCDC

[tidb@localhost ~]$ cat scaleout-cdc.yaml

cdc_servers:

  - host: 192.168.135.148

    gc-ttl: 86400

    data_dir: /tidb-data/cdc-8300

##    gc-ttl: 86400 TiCDC 在 PD 设置的服务级别 GC safepoint 的 TTL (Time To Live) 时长，和 TiCDC 同步任务所能够停滞的时长。单位为秒，默认值为 86400，即 24 小时。

[tidb@localhost ~]$ tiup cluster scale-out  tidb-jiantest scaleout-cdc.yaml

[tidb@localhost ~]$ tiup cluster display tidb-jiantest

192.168.135.148:8300   cdc           192.168.135.148  8300                             linux/x86_64  Up       /tidb-data/cdc-8300           /tidb-deploy/cdc-8300

以下的实践是基于TiDB--TiCDC--MariaDB的价基础上，另外它的下游还可以是Kafka这样就给数据的复制带来了灵活性，一些异构的数据库通过kafka的数据流都可以进行复制Tidb的数据。Confluent Debezium都提供了许多sink的插件，我们也可以自己进行开发。如果数据写入kafka，除了提供默认的开放数据协议 (TiCDC Open Protocol)外，还有多种 kafka 消息协议可选择比如canal、canal-json、avro、maxwell等方式，兼容之前MySQL的cdc，可以无缝迁移。在本文末尾也简单实现可TiCDC到kafka数据同步的演示。

2 创建同步任务

这里要保证两端数据是一致的可以实现使用dumpling工具将两端的数据初始化做好

cdc cli changefeed create --pd=http://192.168.135.148:2379 --sink-uri="mysql://jian:123456@192.168.135.148:3306/" --changefeed-id="tidb-replication-mariadb" --sort-engine="unified"

 以下是几个注意的地方

1 在创建cdc之前，已经存在数据，cdc将其判断为“没有资格的表”会询问将其忽略

2 在两个库中的数据是一致的这样的是没有问题的，报错中没有提示说jian2这张表有问题

3 在目标端是不存在的表也是被cdc判断为“没有资格的表”会询问将其忽略

4 没有主键也是不可以的，这里后边还有两个关于错误创建表的例子

5 关于时区我们也需要将两边调整为一致

调整时区这里以中国时区为例

os端

[root@localhost ~]# ln -s /usr/share/zoneinfo/Asia/Shanghai /etc/localtime

tidb端

MySQL [(none)]> set  global time_zone = Asia/Shanghai;

maridb端

mysql_tzinfo_to_sql /usr/share/zoneinfo | mysql -u root mysql -p

///

在启动changefeed时也可以手动指定

me_zone=CST" --changefeed-id="tidb-replication-mariadb2" --sort-engine="unified"

///

再次启动可以发现已经没有问题了

[tidb@localhost ~]$ cdc cli changefeed create --pd=http://192.168.135.148:2379 --sink-uri="mysql://jian:123456@192.168.135.148:3306/" --changefeed-id="tidb-replication-mariadb" --sort-engine="unified"

Create changefeed successfully!

ID: tidb-replication-mariadb

Info: {"sink-uri":"mysql://jian:123456@192.168.135.148:3306/","opts":{"_changefeed_id":"sink-verify"},"create-time":"2022-03-17T10:57:15.493198043+08:00","start-ts":431878526849777668,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null},"mounter":{"worker-num":16},"sink":{"dispatchers":null,"protocol":"","column-selectors":null},"cyclic-replication":{"enable":false,"replica-id":0,"filter-replica-ids":null,"id-buckets":0,"sync-ddl":false},"scheduler":{"type":"table-number","polling-time":-1},"consistent":{"level":"none","max-log-size":64,"flush-interval":1000,"storage":""}},"state":"normal","error":null,"sync-point-enabled":false,"sync-point-interval":600000000000,"creator-version":"v5.4.0"}

3 简单的看一下数据同步的效果

1 运行之后再创建没有主键的表也都会被忽略,但是不会引起现有复制中断或者影响

两个创建表的错误示范

2 创建多个指向同一个targetdb的cdc进程的时候插入数据不会有任何错误，ddl会在内部以info的形式显示出重复，但是结果是没有问题的都会正常复制

3 TiCDC从特定时间开始

在某些场景下我们需要跳过一些不需要的事务，或者从某一个特定的时间开始，在TiCDC中我们也是可以做到的

为了模拟这里首先我们先暂停changefeed，暂停之后就不会再将后续的事务进行复制，之后Tidb这一侧我们插入一些数据（大家既可以把它当做需要忽略的数据也可以把它当做正常的数据库事务）然后我们找一个指定的位置开始复制。

我们可以获取到数据库中每一条事务的TSO，根据TSO我们就可以从任意的位置开始我们的复制，详细请看下图

这里我们是新创建了一个changefeed这个样例的话

               1 模拟从指定位置开始是没有问题的

               2 模拟跳过事务由于resume changefeed没有提供start_ts的选项所以我们只能将之前的changefeed remove再创建新的changefeed指定start_ts即可

4 cdc管理工具的一些常用操作

[tidb@localhost ~]$ cdc cli capture list --pd=http://192.168.135.148:2379

[tidb@localhost ~]$ cdc cli changefeed create --pd=http://192.168.135.148:2379 --sink-uri="mysql://jian:123456@192.168.135.148:3306/" --changefeed-id="tidb-replication-mariadb" --sort-engine="unified"

[tidb@localhost ~]$ cdc cli changef

[tidb@localhost ~]$ cdc cli changefeed resume  --changefeed-id="tidb-replication-mariadb"

[tidb@localhost ~]$ cdc cli changefeed remove--changefeed-id="tidb-replication-mariadb"

[tidb@localhost ~]$ cdc cli changefeed list --pd=http://192.168.135.148:2379

[tidb@localhost ~]$ cdc cli changefeed query -s --pd=http://192.168.135.148:2379 --changefeed-id=tidb-replication-mariadb

同步子任务处理单元查看

[tidb@localhost ~]$ cdc cli processor list --pd=http://192.168.135.148:2379

[tidb@localhost ~]$ cdc cli processor query --pd=http://192.168.135.148:2379 --changefeed-id=tidb-replication-mariadb --capture-id=77dc637b-1512-4f45-a975-026fb66bae48

5TiCDC 到Kafka的同步

一下只是一个简单的演示

[tidb@localhost ~]$ cdc cli changefeed create --pd="http://192.168.135.148:2379" --sink-uri="" --changefeed-id="tidb-replication-kafka" --opts "registry=http://192.168.135.148:8084"

[tidb@localhost ~]$ cdc cli changefeed list --pd=http://192.168.135.148:2379

[

  {

    "id": "tidb-replication-kafka",

    "summary": {

"state": "normal",

      "tso": 431899473179312130,

      "checkpoint": "2022-03-17 21:08:59.362",

      "error": null

    }

  },

1 使用consumer查看schema中的表结构定义

[root@localhost bin]# /kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.135.148:9092 --topic _schemas --from-beginning

2 使用consumer查看数据库对应的topic中的内容

[root@localhost conf]#/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.135.148:9092 --topic jian --from-beginning

在Tidb端不断插入数据是可以一直看到jian的topic中有输出的但是输出的内容并不是很友好，因为Tidb传入的是avro之后的数据。