数据传输服务 Avro Demo 说明（Flink）

一、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：协议定义文件。

{ "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

数据传输服务 Avro Demo 说明（Flink）

一、Demo 关键逻辑讲解

二、数据库字段映射关系

相关内容：