Avro Demo 说明(Flink)

最近更新时间:2024-03-18 10:41:21

我的收藏

Demo 关键逻辑讲解

Demo 中的文件说明如下:
consumerDemo-avro-flink\\src\\main\\resources\\avro-tools-1.8.2.jar:用来生成 Avro 协议相关代码的工具。
consumerDemo-avro-flink\\src\\main\\java\\com\\tencent\\subscribe\\avro:Avro 工具生成代码的目录。
consumerDemo-avro-flink\\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"
]
}
},
{
……
},
}
Record 中的字段类型解释如下:
Record 中的字段名称
说明
id
全局递增 ID。
version
协议版本,当前版本为1。
messageType
消息类型,枚举值:"INSERT","UPDATE","DELETE","DDL","BEGIN","COMMIT","HEARTBEAT","CHECKPOINT"。
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