Avro Demo 说明

最近更新时间：2024-02-02 16:19:51

我的收藏

本页目录：

Demo 关键逻辑讲解
数据库字段映射关系

Demo 关键逻辑讲解
Demo 中的文件说明如下，以 Java Demo 为例进行介绍。
consumerDemo-avro-java\\src\\main\\resources\\avro-tools-1.8.2.jar：用来生成 Avro 协议相关代码的工具。
consumerDemo-avro-java\\src\\main\\java\\com\\tencent\\subscribe\\avro：Avro 工具生成代码的目录。
consumerDemo-avro-java\\src\\main\\resources\\Record.avsc：协议定义文件。
Record.avsc 中我们定义了14个结构（Avro 中叫做 schema），其中主要的数据结构为 Record，用于表示 binlog 中的一条数据，Record 的结构如下，其他数据结构可以在 Record.avsc 中查看：
 {
    "namespace": "com.tencent.subscribe.avro",    //Record.avsc 中的最后1个 schema，"name" 显示为 "Record"
    "type": "record",    
    "name": "Record",     //"name" 显示为 "Record"，表示从 kafka 中消费的数据格式
    "fields": [
      {
        "name": "id",     //id 表示全局递增 ID，更多 record 取值解释如下表
        "type": "long",
        "doc": "unique id of this record in the whole stream"
      },
      {
        "name": "version",  //version 表示协议版本
        "type": "int",
        "doc": "protocol version"  
      },
      {
        "name": "messageType",   //消息类型
        "aliases": [
          "operation"
        ],
        "type": {
          "namespace": "com.tencent.subscribe.avro",
          "name": "MessageType",
          "type": "enum",
          "symbols": [
            "INSERT",
            "UPDATE",
            "DELETE",
            "DDL",
            "BEGIN",
            "COMMIT",
            "HEARTBEAT",
            "CHECKPOINT",
            "ROLLBACK"          
            "STATEMENT",         
            "INIT_DDL",          
            "INIT_INSERT"
          ]
        }
      },  
      {
       ……
      },     
 }    
Record 中的字段类型解释如下：
Record 中的字段名称
说明
id
全局递增 ID。
version
协议版本，当前版本为1。
messageType
消息类型，枚举值："INSERT"，"UPDATE"，"DELETE"，"DDL"，"BEGIN"，"COMMIT"，"HEARTBEAT"，"CHECKPOINT"，"STATEMENT"，"INIT_DDL"，"INIT_INSERT"。
其中，"INIT_DDL"表示全量数据的 DDL，"DDL"表示增量数据的 DDL，"INIT_INSERT"表示全量数据的 INSERT，"INSERT"表示增量数据的 INSERT。
fileName
当前 record 所在的 binlog 文件名。
position
当前 record 的在 binlog 中结束的偏移量，格式为 End_log_pos@binlog 文件编号。例如，当前 record 位于文件 mysql-bin.000004 中，结束偏移量为2196，则其值为"2196@4"。
safePosition
当前事务在 binlog 中开始的偏移量，格式同上。
timestamp
写 binlog 的时间，unix 时间戳，秒级。
gtid
当前的 gtid，如：c7c98333-6006-11ed-bfc9-b8cef6e1a231:9。
transactionId
事务 ID，只有 commit 事件才会生成事务 ID。
serverId
源库 serverId，查看源库 server_id 参考 SHOW VARIABLES LIKE 'server_id'。
threadId
提交当前事务的会话 ID，参考 SHOW processlist;。
sourceType
源库的数据库类型，当前版本只有 MySQL。
sourceVersion
源库版本，查看源库版本参考select version();。
schemaName
库名。
tableName
表名。
objectName
格式为：库名.表名。
columns
表中各列的定义。
oldColumns
DML 执行前该行的数据，如果是 insert 消息，该数组为 null。数组中元素共有12种类型：Integer，Character，Decimal，Float，Timestamp，DateTime，TimestampWithTimeZone，BinaryGeometry，TextGeometry，BinaryObject，TextObject，EmptyObject，详见 demo 中定义。
newColumns
DML 执行后该行的数据，如果是 delete 消息，该数组为 null。数组中元素共有12种类型：Integer，Character，Decimal，Float，Timestamp，DateTime，TimestampWithTimeZone，BinaryGeometry，TextGeometry，BinaryObject，TextObject，EmptyObject，详见 demo 中定义。
sql
DDL 的 SQL 语句。
executionTime
DDL 执行时长，单位为秒。
heartbeatTimestamp
心跳消息的时间戳，秒级。只有 heartbeat 消息才有该字段。
syncedGtid
DTS 已解析 GTID 集合，格式形如：c7c98333-6006-11ed-bfc9-b8cef6e1a231:1-13。
fakeGtid
是否为构造的假 GTID，如未开启 gtid_mode，则 DTS 会构造一个 GTID。
pkNames
如果源库的表设有主键，则 DML 消息中会携带该参数，否则不会携带。
readerTimestamp
DTS 处理这条数据的时间，unix 时间戳，单位为毫秒数。
tags
QueryEvent 中的 status_vars，详细参考 QueryEvent。
total
如果消息分片，记录分片总数。当前版本 (version=1) 无意义，预留扩展。
index
如果消息分片，记录当前分片的索引。当前版本 (version=1) 无意义，预留扩展。
Record 中描述列属性的字段为 "Field"，包含如下四个属性：
name：列名。
dataTypeNumber：是 binlog 中记录的数据类型。取值详见 MySQL。
isKey：是否主键。
originalType：DDL 中定义的类型。
数据库字段映射关系
如下为数据库（如 MySQL）字段类型和 Avro 协议中定义的数据类型之间的映射关系。 
MySQL 类型
对应 Avro 中的类型
MYSQL_TYPE_NULL
EmptyObject
MYSQL_TYPE_INT8
Integer
MYSQL_TYPE_INT16
Integer
MYSQL_TYPE_INT24
Integer
MYSQL_TYPE_INT32
Integer
MYSQL_TYPE_INT64
Integer
MYSQL_TYPE_BIT
Integer
MYSQL_TYPE_YEAR
DateTime
MYSQL_TYPE_FLOAT
Float
MYSQL_TYPE_DOUBLE
Float
MYSQL_TYPE_VARCHAR
Character
MYSQL_TYPE_STRING
Character，如果原类型为 binary，则对应 BinaryObject
MYSQL_TYPE_VAR_STRING
Character，如果原类型为 varbinary，则对应 BinaryObject
MYSQL_TYPE_TIMESTAMP
Timestamp
MYSQL_TYPE_DATE
DateTime
MYSQL_TYPE_TIME
DateTime
MYSQL_TYPE_DATETIME
DateTime
MYSQL_TYPE_TIMESTAMP_NEW
Timestamp
MYSQL_TYPE_DATE_NEW
DateTime
MYSQL_TYPE_TIME_NEW
DateTime
MYSQL_TYPE_DATETIME_NEW
DateTime
MYSQL_TYPE_ENUM
TextObject
MYSQL_TYPE_SET
TextObject
MYSQL_TYPE_DECIMAL
Decimal
MYSQL_TYPE_DECIMAL_NEW
Decimal
MYSQL_TYPE_JSON
TextObject
MYSQL_TYPE_BLOB
BinaryObject
MYSQL_TYPE_TINY_BLOB
BinaryObject
MYSQL_TYPE_MEDIUM_BLOB
BinaryObject
MYSQL_TYPE_LONG_BLOB
BinaryObject
MYSQL_TYPE_GEOMETRY
BinaryObject
﻿

上一篇: 数据消费操作指导下一篇: JSON Demo 说明

Record 中的字段名称	说明
id	全局递增 ID。
version	协议版本，当前版本为1。
messageType	消息类型，枚举值："INSERT"，"UPDATE"，"DELETE"，"DDL"，"BEGIN"，"COMMIT"，"HEARTBEAT"，"CHECKPOINT"，"STATEMENT"，"INIT_DDL"，"INIT_INSERT"。其中，"INIT_DDL"表示全量数据的 DDL，"DDL"表示增量数据的 DDL，"INIT_INSERT"表示全量数据的 INSERT，"INSERT"表示增量数据的 INSERT。
fileName	当前 record 所在的 binlog 文件名。
position	当前 record 的在 binlog 中结束的偏移量，格式为 End_log_pos@binlog 文件编号。例如，当前 record 位于文件 mysql-bin.000004 中，结束偏移量为2196，则其值为"2196@4"。
safePosition	当前事务在 binlog 中开始的偏移量，格式同上。
timestamp	写 binlog 的时间，unix 时间戳，秒级。
gtid	当前的 gtid，如：c7c98333-6006-11ed-bfc9-b8cef6e1a231:9。
transactionId	事务 ID，只有 commit 事件才会生成事务 ID。
serverId	源库 serverId，查看源库 server_id 参考 SHOW VARIABLES LIKE 'server_id'。
threadId	提交当前事务的会话 ID，参考 SHOW processlist;。
sourceType	源库的数据库类型，当前版本只有 MySQL。
sourceVersion	源库版本，查看源库版本参考`select version();`。
schemaName	库名。
tableName	表名。
objectName	格式为：库名.表名。
columns	表中各列的定义。
oldColumns	DML 执行前该行的数据，如果是 insert 消息，该数组为 null。数组中元素共有12种类型：Integer，Character，Decimal，Float，Timestamp，DateTime，TimestampWithTimeZone，BinaryGeometry，TextGeometry，BinaryObject，TextObject，EmptyObject，详见 demo 中定义。
newColumns	DML 执行后该行的数据，如果是 delete 消息，该数组为 null。数组中元素共有12种类型：Integer，Character，Decimal，Float，Timestamp，DateTime，TimestampWithTimeZone，BinaryGeometry，TextGeometry，BinaryObject，TextObject，EmptyObject，详见 demo 中定义。
sql	DDL 的 SQL 语句。
executionTime	DDL 执行时长，单位为秒。
heartbeatTimestamp	心跳消息的时间戳，秒级。只有 heartbeat 消息才有该字段。
syncedGtid	DTS 已解析 GTID 集合，格式形如：c7c98333-6006-11ed-bfc9-b8cef6e1a231:1-13。
fakeGtid	是否为构造的假 GTID，如未开启 gtid_mode，则 DTS 会构造一个 GTID。
pkNames	如果源库的表设有主键，则 DML 消息中会携带该参数，否则不会携带。
readerTimestamp	DTS 处理这条数据的时间，unix 时间戳，单位为毫秒数。
tags	QueryEvent 中的 status_vars，详细参考 QueryEvent。
total	如果消息分片，记录分片总数。当前版本 (version=1) 无意义，预留扩展。
index	如果消息分片，记录当前分片的索引。当前版本 (version=1) 无意义，预留扩展。

MySQL 类型	对应 Avro 中的类型
MYSQL_TYPE_NULL	EmptyObject
MYSQL_TYPE_INT8	Integer
MYSQL_TYPE_INT16	Integer
MYSQL_TYPE_INT24	Integer
MYSQL_TYPE_INT32	Integer
MYSQL_TYPE_INT64	Integer
MYSQL_TYPE_BIT	Integer
MYSQL_TYPE_YEAR	DateTime
MYSQL_TYPE_FLOAT	Float
MYSQL_TYPE_DOUBLE	Float
MYSQL_TYPE_VARCHAR	Character
MYSQL_TYPE_STRING	Character，如果原类型为 binary，则对应 BinaryObject
MYSQL_TYPE_VAR_STRING	Character，如果原类型为 varbinary，则对应 BinaryObject
MYSQL_TYPE_TIMESTAMP	Timestamp
MYSQL_TYPE_DATE	DateTime
MYSQL_TYPE_TIME	DateTime
MYSQL_TYPE_DATETIME	DateTime
MYSQL_TYPE_TIMESTAMP_NEW	Timestamp
MYSQL_TYPE_DATE_NEW	DateTime
MYSQL_TYPE_TIME_NEW	DateTime
MYSQL_TYPE_DATETIME_NEW	DateTime
MYSQL_TYPE_ENUM	TextObject
MYSQL_TYPE_SET	TextObject
MYSQL_TYPE_DECIMAL	Decimal
MYSQL_TYPE_DECIMAL_NEW	Decimal
MYSQL_TYPE_JSON	TextObject
MYSQL_TYPE_BLOB	BinaryObject
MYSQL_TYPE_TINY_BLOB	BinaryObject
MYSQL_TYPE_MEDIUM_BLOB	BinaryObject
MYSQL_TYPE_LONG_BLOB	BinaryObject
MYSQL_TYPE_GEOMETRY	BinaryObject