TiDB TiCDC 使用实践

作者： banana_jian 原文来源：https://tidb.net/blog/6c5d6482

   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 时也可以手动指定

[tidb\@localhost ~]$ cdc cli changefeed create –pd=http://192.168.135.148:2379 –sink-uri=“mysql://jian:123456\@192.168.135.148:3306/?time_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 changefeed update –pd=http://192.168.135.148:2379 –sink-uri=“mysql://jian:123456\@192.168.135.148:3306/?max-txn-row=20&worker-number=8” –changefeed-id=“tidb-replication-mariadb” –sort-engine=“unified”

[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=”kafka://192.168.135.148:9092/jian?protocol=avro” –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 之后的数据。