腾讯云消息队列 CKafka - MySQL 订阅消息 canal 格式说明
文档简介：
概述：
使用连接器订阅 MySQL 的变更操作时，可选多种消息格式，默认采用 debezium 格式，同时提供了兼容开源 canal 等格式的能力。本文介绍兼容 canal 的消息格式说明。
咨询热线：400-826-7010，为您提供专业的售前咨询，让您快速了解云产品，助您轻松上云！
微信咨询
免费试用、价格特惠
概述
使用连接器订阅 MySQL 的变更操作时，可选多种消息格式，默认采用 debezium 格式，同时提供了兼容开源 canal 等格式的能力。本文介绍兼容 canal 的消息格式说明。
canal 格式说明
|
兼容字段
|
canal原生字段
|
是否支持
|
|
id（默认值0）
|
id（canal 生成的消息 ID）
|
否
|
|
database(数据库名)
|
database(数据库名)
|
是
|
|
table（表名）
|
table（表名）
|
是
|
|
pkNames（默认值null）
|
pkNames（主键字段名）
|
否
|
|
isDdl（变更是否属于DDL）
|
isDdl（变更是否属于 DDL）
|
是
|
|
type（变更类型）
|
type（变更类型）
|
是
|
|
es（binlog时间戳）
|
es（binlog 时间戳）
|
是
|
|
ts（connector 同步时间，仅 DML 支持，DDL 中与 ts 一致）
|
ts（connector 同步时间）
|
是
|
|
sql（DDL 执行语句）
|
sql（DDL 执行语句）
|
是
|
|
sqlType（暂不支持，默认值 null）
|
sqlType（与 mysqlType 对应的数据类型编号）
|
否
|
|
mysqlType（暂不支持，默认值 null）
|
mysqlType（mysql 中的字段类型）
|
否
|
|
data（变更后的记录-全字段）
|
data（变更后的记录-全字段）
|
是
|
|
old（变更前的记录--全字段）
|
old（变更前的记录--仅变更字段）
|
是
|
DDL 格式
create database
{"data": null,"database": "dip_test","es": 1655812326,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "CREATE DATABASE `dip_test` CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci","sqlType": null,"table": "","ts": 1655812326,"type": "QUERY"}
drop database
{"data": null,"database": "dip_test","es": 1655812326,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "DROP DATABASE IF EXISTS `dip_test`","sqlType": null,"table": "","ts": 1655812326,"type": "QUERY"}
create table
{"data": null,"database": "dip_test","es": 1655812326,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "CREATE TABLE `customers` (`id` int NOT NULL AUTO_INCREMENT,`first_name` varchar(255) NOT NULL,`last_name` varchar(255) NOT NULL,`email` varchar(255) NOT NULL,PRIMARY KEY (`id`),UNIQUE KEY `email` (`email`),KEY `ix_id` (`id`)) ENGINE=InnoDB AUTO_INCREMENT=1041 DEFAULT CHARSET=utf8","sqlType": null,"table": "customers","ts": 1655812326,"type": "CREATE"}
alter table
{"data": null,"database": "test","es": 1655782153,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "ALTER TABLE `user` ADD COLUMN `createtime` datetime NULL DEFAULT CURRENT_TIMESTAMP","sqlType": null,"table": "user","ts": 1655782153,"type": "ALTER"}
drop table
{"data": null,"database": "dip_test","es": 1655812326,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "DROP TABLE IF EXISTS `dip_test`.`customers`","sqlType": null,"table": "customers","ts": 1655812326,"type": "ERASE"}
rename table
{"data": null,"database": "testDB","es": 1656300979748,"id": 0,"isDdl": true,"mysqlType": null,"old": null,"pkNames": null,"sql": "rename table test to t_test","sqlType": null,"table": "t_test","ts": 1656300979748,"type": "RENAME"}
DML 格式
insert
{"data": [{"last_name": "Kretchmar","id": 1004,"first_name": "Anne","email": "annek@noanswer.org"}],"database": "inventory","es": 0,"id": 0,"isDdl": false,"mysqlType": null,"old": null,"pkNames": null,"sql": "","sqlType": null,"table": "customers","ts": 1465491411815,"type": "INSERT"}
update
{"data": [{"last_name": "Kretchmar","id": 1004,"first_name": "Anne Marie","email": "annek@noanswer.org"}],"database": "inventory","es": 1465581029100,"id": 0,"isDdl": false,"mysqlType": null,"old": [{"last_name": "Kretchmar","id": 1004,"first_name": "Anne","email": "annek@noanswer.org"}],"pkNames": null,"sql": "","sqlType": null,"table": "customers","ts": 1465581029523,"type": "UPDATE"}
delete
{"data": null,"database": "inventory","es": 1465581902300,"id": 0,"isDdl": false,"mysqlType": null,"old": [{"last_name": "Kretchmar","id": 1004,"first_name": "Anne Marie","email": "annek@noanswer.org"}
