1. 首页
  2. 文章中心
  3. json

TiCDC canal_json的实际应用 TiCDC Canal-JSON

68 阅读 0 评论
我是靠谱客的博主 傻傻导师，最近开发中收集的这篇文章主要介绍TiCDC canal_json的实际应用 TiCDC Canal-JSON ，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

作者： Jiawei 原文来源： https://tidb.net/blog/bc8452e1

背景知识

在开始介绍之前先和大家简单介绍两个东西：

1 .目前流行的` 缓存和DB一致性 `的实现架构： no-alt

基本的流程如图所示：

MySQL 增删改 --> Canal(伪装slave)获取变更--> kafka 接收topic 写入 --> api消费kafka topic 获取变更 -->失效缓存

2 . Canal

Canal是阿里早期为了解决异国双机房之间数据同步业务需求而开发出来的基于日志解析和变更进行同步的工具，由此衍生出了大量的数据库增量订阅和消费业务。

官方文档地址： otter

架构图：

no-alt

简单讲canal就是伪装成MySQL的slave，然后像主从同步的样子，去获取变更。

canal的使用场景一般也有2个种类：

1.缓存DB一致性实现。

2.同步数据到数仓平台。

TiCDC Canal-JSON

Canal-JSON 是什么？

Canal-JSON 其实是阿里巴巴定义的一种数据交换格式协议，本身就是为MySQL设计的。

为什么要用到TiCDC Canal-JSON协议？

原因是因为我们之前数据库是在MySQL上面跑的，然后近期MySQL迁移到了TiDB上面，为了保证之前那一套缓存一致性业务逻辑实现，而canal又不支持TiDB，所以我们需要一个可以替换canal的组件来获取TiDB的变更并写入到后端的Message Queue。

经过我的调研，TiDB有两种方式可以实现我们的需求：

1.是用TiDB binlog工具可以将变更写到 MQ。

2.是用TiCDC创建一个同步任务，指定数据格式是Canal-JSON样式。

而且经过查看TICDC canal_json 和canal实现对比如下，更多可以看官方文档： TiCDC Canal-JSON Protocol

不同点 TiCDC Canal
update old块包含老的所有列的原数据 old块只包含被更新的列的数据
data type 没有类型参数，比如char(16) 只显示char 而canal会显示char(16)
commit_ts唯一标识 sink_uri enable-tidb-extension 开启会显示 多一个_tidb字段显示commit_ts 无
delete type v5.4之前，old和data内容相同 v5.4之后，old为null 无

所以经过上面的调研，我们最终决定使用第二种TiCDC的方式，原因2点：

  • canal和TiCDC canal_json 大部分实现相同，差异很小。

  • 对于开发友好，我们的开发之前都是对canal比较熟悉，可以很方便快速上手，改动更少的代码。

TiCDC canal_json 如何使用？

首先非常关键的一点，canal_json还是机遇TiCDC实现的，所以我们肯定得先有TiCDC集群。

以下操作基于TiDB 6.0版本。

1.首先创建一个配置文件

创建配置文件的原因是因为我们实际并不是需要某个业务库下的所有表，只有一些重要的热点的表。

因为canal可以在配置文件中配置白名单黑名单，如下 

# 白名单，表示是dbname数据库的所有表
canal.instance.filter.regex=dbname\..*
# 黑名单，避免同步无用的增量数据
canal.instance.filter.black.regex=

TiCDC没法直接指定我们需要同步那些库的哪些表，所以我们需要在配置文件指定 

#比如这里我们只要同步这台TiDB集群的ticdc_canal_test数据库的表，其他都不同步
$ cat /root/ticdc_canal_test.toml
[filter]
rules = ['ticdc_canal_test.*']

这里只是简单演示下，更详细的过滤规则大家可以参考官方文档: 配置文件配置

2.创建一个同步任务

配置好了之后我们可以开始配置同步任务，先了解一些常用参数： 

--changefeed-id          任务ID
--sink-uri               同步任务下游的地址，目前支持mysql/tidb/kafka/pulsar
--start-ts               同步任务开始的位置，这里是ts，默认创建任务当前时间，类似于mysql搭建同步的开始文件位点
--target-ts              同步任务结束位置，默认为空，即一直同步
--config                 指定配置文件，过滤规则这些

知道这些含义之后我们可以开始创建一个任务： 

#1.创建一个同步到kafka的同步任务
$ cdc cli changefeed create --pd=http://pd_ip:42379 --changefeed-id="ticdc-canal-json-test" --config="/root/ticdc_canal_test.toml" --sink-uri="kafka://kafka_ip:9092/ticdc_canal_test_topic?kafka-version=2.6.0&protocol=canal-json&enable-tidb-extension=true"
[2022/06/02 16:23:27.994 +08:00] [WARN] [kafka.go:451] ["broker's `message.max.bytes` less than the `max-message-bytes`,use broker's `message.max.bytes` to initialize the Kafka producer"] [message.max.bytes=1048588] [max-message-bytes=10485760]
[2022/06/02 16:23:27.994 +08:00] [WARN] [kafka.go:461] ["partition-num is not set, use the default partition count"] [topic=ticdc_canal_test_topic] [partitions=3]
Create changefeed successfully!
ID: ticdc-canal-json-test
Info: {"sink-uri":"kafka://kafka_ip:9092/ticdc_canal_test_topic?kafka-version=2.6.0u0026protocol=canal-jsonu0026enable-tidb-extension=true","opts":{},"create-time":"2022-06-02T16:23:27.974959914+08:00","start-ts":433627649254096897,"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":["ticdc_canal_test.*"],"ignore-txn-start-ts":null},"mounter":{"worker-num":16},"sink":{"dispatchers":null,"protocol":"canal-json","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":"v6.0.0"}

#2.查看刚才的任务状态,可以看到详细的信息
$ cdc cli changefeed query --pd=http://pd_ip:42379 --changefeed-id="ticdc-canal-json-test"
{
  "info": {
    "sink-uri": "kafka://kafka_ip:9092/ticdc_canal_test_topic?kafka-version=2.6.0u0026protocol=canal-jsonu0026enable-tidb-extension=true",
    "opts": {},
    "create-time": "2022-06-02T16:23:27.974959914+08:00",
    "start-ts": 433627649254096897,
    "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": [
          "ticdc_canal_test.*"
        ],
        "ignore-txn-start-ts": null
      },
      "mounter": {
        "worker-num": 16
      },
      "sink": {
        "dispatchers": null,
        "protocol": "canal-json",
        "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": "v6.0.0"
  },
  "status": {
    "resolved-ts": 433627686504759297,
    "checkpoint-ts": 433627686504759297,
    "admin-job-type": 0
  },
  "count": 0,
  "task-status": [
    {
      "capture-id": "48f5f942-025d-48c3-a7c3-e06c70334ef2",
      "status": {
        "tables": null,
        "operation": null,
        "admin-job-type": 0
      }
    }
  ]
}

3.数据库操作变更 

MySQL [(none)]> use ticdc_canal_test
Database changed
MySQL [ticdc_canal_test]> show tables;
Empty set (0.00 sec)

MySQL [ticdc_canal_test]>  CREATE TABLE `jiawei_test2` (
    ->   `id` int(11) NOT NULL,
    ->   `name` varchar(10) COLLATE utf8mb4_general_ci DEFAULT NULL,
    ->   PRIMARY KEY (`id`));
Query OK, 0 rows affected (0.09 sec)

MySQL [ticdc_canal_test]> show create table jiawei_test2
    -> ;
+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Table        | Create Table                                                                                                                                                                                                                             |
+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| jiawei_test2 | CREATE TABLE `jiawei_test2` (
  `id` int(11) NOT NULL,
  `name` varchar(10) COLLATE utf8mb4_general_ci DEFAULT NULL,
  PRIMARY KEY (`id`) /*T![clustered_index] CLUSTERED */
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin |
+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

MySQL [ticdc_canal_test]> insert into jiawei_test2(id,name) values(1,'numer1'),(2,'number2'),(3,'number3');
Query OK, 3 rows affected (0.01 sec)
Records: 3  Duplicates: 0  Warnings: 0

MySQL [ticdc_canal_test]> select * from jiawei_test2;
+----+---------+
| id | name    |
+----+---------+
|  1 | numer1  |
|  2 | number2 |
|  3 | number3 |
+----+---------+
3 rows in set (0.01 sec)

MySQL [ticdc_canal_test]> update jiawei_test2 set name='tidb';
Query OK, 3 rows affected (0.01 sec)
Rows matched: 3  Changed: 3  Warnings: 0

MySQL [ticdc_canal_test]> select * from jiawei_test2;
+----+------+
| id | name |
+----+------+
|  1 | tidb |
|  2 | tidb |
|  3 | tidb |
+----+------+
3 rows in set (0.01 sec)

MySQL [ticdc_canal_test]> delete from jiawei_test2 where id=1;
Query OK, 1 row affected (0.01 sec)

MySQL [ticdc_canal_test]> select * from jiawei_test2;
+----+------+
| id | name |
+----+------+
|  2 | tidb |
|  3 | tidb |
+----+------+
2 rows in set (0.00 sec)

MySQL [ticdc_canal_test]> alter table jiawei_test2 add column c1 int;
Query OK, 0 rows affected (0.27 sec)

MySQL [ticdc_canal_test]> select * from jiawei_test2;
+----+------+------+
| id | name | c1   |
+----+------+------+
|  2 | tidb | NULL |
|  3 | tidb | NULL |
+----+------+------+
2 rows in set (0.00 sec)

MySQL [ticdc_canal_test]> alter table jiawei_test2 add index idx_name(name);
Query OK, 0 rows affected (2.80 sec)

MySQL [ticdc_canal_test]> alter table jiawei_test2 drop column c1;
Query OK, 0 rows affected (0.28 sec)

MySQL [ticdc_canal_test]> select * from jiawei_test2;
+----+------+
| id | name |
+----+------+
|  2 | tidb |
|  3 | tidb |
+----+------+
2 rows in set (0.01 sec)

4.查看topic内容 

#测试一下消费信息
$ ./bin/kafka-console-consumer.sh --bootstrap-server kafka_ip:9092 --topic test_ticdc_canal_json --from-beginning
#创建表消息
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":null,"isDdl":true,"type":"CREATE","es":1654163280810,"ts":1654163282102,"sql":"CREATE TABLE `jiawei_test2` (`id` INT(11) NOT NULL,`name` VARCHAR(10) COLLATE utf8mb4_general_ci DEFAULT NULL,PRIMARY KEY(`id`))","sqlType":null,"mysqlType":null,"data":null,"old":null}
#插入数据消息
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1654163303159,"ts":1654163305460,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"1","name":"numer1"}],"old":null}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1654163303159,"ts":1654163305461,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"2","name":"number2"}],"old":null}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1654163303159,"ts":1654163305461,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"3","name":"number3"}],"old":null}
#更新数据消息，可以看到old是包含了所有列原来的值
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"UPDATE","es":1654163312959,"ts":1654163314469,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"1","name":"tidb"}],"old":[{"id":"1","name":"numer1"}]}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"UPDATE","es":1654163312959,"ts":1654163314469,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"2","name":"tidb"}],"old":[{"id":"2","name":"number2"}]}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"UPDATE","es":1654163312959,"ts":1654163314470,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"3","name":"tidb"}],"old":[{"id":"3","name":"number3"}]}
#删除数据信息，可以看到确实6.0的版本old是null
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":["id"],"isDdl":false,"type":"DELETE","es":1654163329309,"ts":1654163330485,"sql":"","sqlType":{"id":4,"name":12},"mysqlType":{"id":"int","name":"varchar"},"data":[{"id":"1","name":"tidb"}],"old":null}
#DDL测试，增加索引，增删列
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":null,"isDdl":true,"type":"ALTER","es":1654163338309,"ts":1654163340302,"sql":"ALTER TABLE `jiawei_test2` ADD COLUMN `c1` INT","sqlType":null,"mysqlType":null,"data":null,"old":null}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":null,"isDdl":true,"type":"CINDEX","es":1654163357509,"ts":1654163360101,"sql":"ALTER TABLE `jiawei_test2` ADD INDEX `idx_name`(`name`)","sqlType":null,"mysqlType":null,"data":null,"old":null}
{"id":0,"database":"ticdc_canal_json_test","table":"jiawei_test2","pkNames":null,"isDdl":true,"type":"ALTER","es":1654163366509,"ts":1654163369502,"sql":"ALTER TABLE `jiawei_test2` DROP COLUMN `c1`","sqlType":null,"mysqlType":null,"data":null,"old":null}

注意事项

建议在同步的时候不要打开 enable-tidb-extension 这个选项，如果这个额外字段不是必须的话，因为这个开启之后会生成很多的 WATERMARK Event ，不方便我们观察和消费

no-alt

总结

目前我们已经在项目上跑了很久了，非常稳定，这里给大家简单介绍了一下基本用法，

想更多的分区等操作，canal支持的，TiCDC canal_json 基本都支持。

跟过的可以参考官方文档更详细参数设置: 参数设置

最后

以上就是傻傻导师为你收集整理的TiCDC canal_json的实际应用 TiCDC Canal-JSON 的全部内容，希望文章能够帮你解决TiCDC canal_json的实际应用 TiCDC Canal-JSON 所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(45)

相关文章

Galera Cluster for Mysql集群环境搭建1、系统环境2、linux 更换国内镜像3、安装清华大学 elep repo5、卸载 mariadb6、开启防火墙7、设置SELinux8、下载 Galear 组件9、安装Galear 依赖包10、安装Galear11、创建wsrep.cnf 文件12、初始化mysql13、注意事项14、启动集群15、服务器重启
Galera Cluster for Mysql集群环境搭建1、系统环境2、linux 更换国内镜像3、安装清华大学 elep repo5、卸载 mariadb6、开启防火墙7、设置SELinux8、下载 Galear 组件9、安装Galear 依赖包10、安装Galear11、创建wsrep.cnf 文件12、初始化mysql13、注意事项14、启动集群15、服务器重启
flink连接kafka，postgresql sink
flink连接kafka，postgresql sink
一键启动Debezium+Kafka+Flink编写配置文件docker-compose.yml将kafka对外开放，方便调试启动
一键启动Debezium+Kafka+Flink编写配置文件docker-compose.yml将kafka对外开放，方便调试启动
flink整合kafka&Debezium(一）项目启动Debezium的搭建产生数据Flink的搭建开发flink项目
flink整合kafka&Debezium(一）项目启动Debezium的搭建产生数据Flink的搭建开发flink项目
TiCDC canal_json的实际应用 TiCDC Canal-JSON
TiCDC canal_json的实际应用 TiCDC Canal-JSON
Flink cdc 读取 kafka json 数据入PG数据库
Flink cdc 读取 kafka json 数据入PG数据库
flink metric对接openTSDB背景方案代码
flink metric对接openTSDB背景方案代码
11.5.2、Canal__json解析
11.5.2、Canal__json解析

评论列表共有 0 条评论

立即
投稿
返回
顶部