--- title: "数据格式说明 | OceanBase 文档中心" description: 数据格式说明 本文为您介绍序列化方式和数据库传输到文本协议的数据格式。 序列化方式的格式说明 您在使用 OceanBase 迁移服务（OceanBase Migration Service，OMS）同步源端数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，支持序列化方式控制数据同步至目标端… --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*L03BS6f-o40AAAAAAAAAAAAADiGDAQ/original) 迁移服务 OMSV 4.3.1 企业版 # 数据格式说明 更新时间：2026-04-17 18:35:39 本文为您介绍序列化方式和数据库传输到文本协议的数据格式。 ## 序列化方式的格式说明 您在使用 OceanBase 迁移服务（OceanBase Migration Service，OMS）同步源端数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，支持序列化方式控制数据同步至目标端的消息格式。序列化方式包括 **Default**、**Canal**、**DataWorks**（支持 V2.0）、**SharePlex**、**DefaultExtendColumnType**、**Debezium**、**DebeziumFlatten**、**DebeziumSmt** 和 **Avro**。 #### 说明 - 目前仅 OceanBase 数据库 MySQL 兼容模式支持序列化方式 `Debezium`、`DebeziumFlatten` 和 `DebeziumSmt`。 - 目前仅同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka 时，支持序列化方式 `Avro`。 ### Default JSON 消息格式 数据同步至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 Default 使用如下 JSON 消息格式。 ```JSON { "prevStruct": { // 变更前镜像 "col1": "val1" // 键值对，包含全量键值 }, "postStruct": { // 变更后镜像 "col1": "val1" // 键值对，包含全量键值 }, "allMetaData"{ "checkpoint": "STRING", // 当前同步位点，增量阶段表示同步到的时间位点（秒级时间戳），全量阶段使用主键键值对表示 "record_primary_key": "STRING", // 主键列的名称。如果存在多列使用 \u0001 分割 "record_primary_value": "STRING", // 主键值。如果存在多列使用 \u0001 分割 "source_identity": "STRING", // 源端标识，如果是增量则是 subtopic，如果是全量则没有意义的序号 "dbType": "STRING", // 数据库的类型。包括 MYSQL/ORACLE/OCEANBASE（老模式，兼容使用）/OB_IN_ORACLE_MODE（老模式，兼容使用）/DB2（老模式，兼容使用）/OB_MYSQL/OB_ORACLE/DB2_LUW "storeDataSequence": "LONG", // 该字段只有在增量场景下 source.json 配置中包含 sequenceEnabled=true 才存在，默认是 true。用于排序，生成规则是一个同步进程中，时间戳 + 不超过五位序号递增。{时间戳}{递增序号}。 "table_name": "STRING", // 使用 SQL 语句进行变更的表的名称 "db": "STRING", // 使用 SQL 语句进行变更的数据库的名称。如果是 OceanBase 数据库，则包含租户，格式为 {tenant}.{database} "timestamp": "STRING", // 数据变更秒级时间戳，仅增量存在。完整的时间戳信息被拆成两部分表示：timestamp 和 nanos。比如 timestamp=1751854603，nanos=123456000，则完整的时间戳为 1751854603 * 1000000000 + 123456000 = 1751854603123456000 ns。 "nanos": "STRING", // 数据变更纳秒级时间戳，仅增量存在 "uniqueId": "STRING", // 增量中表示 STORE 传递下来的事务序号标识 "transId": "STRING", // OceanBase 数据库 V4.x 之前版本表示事务 ID（inc 是一个自增数，addr 是协调者地址，ts 是事务 ID 生成时的时间，hash 是上述三个的 hash 值），如果事务不完整则为 null。例如，"transId":"{hash:614827533592190202, inc:1384316437, addr:\"xxx.xxx.xxx.xxx:2882\", t:1756190533588977}"。OceanBase 数据库 V4.x 及之后版本表示租户 ID_事务 ID。例如，"transId":"1xxx_3152xxx"。 "clusterId": "STRING", // 在 OceanBase 数据库中表示 clusterId，在 MySQL 数据库中表示 threadId，在 Oracle 数据库中不存在。 "ddlType": "STRING", // DDL 具体类型，OMS V4.0.1 新增支持 }, "recordType": "INSERT/UPDATE/DELETE/HEARTBEAT/DDL/ROW" // 变更类型 } ``` - DDL 的 Record 中，仅存在 "ddl" 为列名的键，值为 DDL 语句。 - 前镜像和后镜像： - `prevStruct`：表示增量数据的前镜像信息，即 SQL 执行前的数据。 - `postStruct`：表示增量数据的后镜像信息，即 SQL 执行后的数据。 `DELETE` 仅存在 `prevStruct`，`INSERT`和`DDL` 仅存在 `postStruct`，`UPDATE` 同时存在 `prevStruct`和 `postStruct`，`HEARTBEAT`（定期心跳消息）不存在 `postStruct` 和 `postStruct。` - `clusterId` 的详细解释如下： - 在 OceanBase 数据库中，`ob_org_cluster_id` 用于设置 OceanBase 数据库 Session 级别的 CLUSTER_ID，该值会持久化到事务日志中。如果您写数据时设置了 `ob_org_cluster_id`，则以您设置的值为准。否则均是缺省值，仅增量存在。详情请参见 [ob_org_cluster_id](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000000510276)。 - 在 MySQL 数据库中，`clusterId` 表示 MySQL Binlog Event 中的 threadId。MySQL Server 会为每个连接上的线程分配递增的 threadId。 数据示例如下： - INSERT（插入）数据的示例 ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "INSERT", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } } ``` `recordType` 为 `ROW` 时，表示是全量传入的数据，格式与 `INSERT` 相同。 ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "ROW", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } } ``` - UPDATE（更新）数据的示例 ```JSON { "allMetaData": { "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx27, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", }, "recordType": "UPDATE", "postStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } } ``` - DELETE（删除）数据的示例 ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx27, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col16": 1.2222, "col7": 9.99999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345" }, "recordType": "DELETE", "postStruct": null } ``` - DDL 示例 ```SQL ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT "test" COMMENT 'test'; ``` ```JSON { "prevStruct": null, "postStruct": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" }, "allMetaData": { "checkpoint": "1671177057", "dbType": "OB_MYSQL", "storeDataSequence": null, "db": "connector_test", "timestamp": "1671177057", "uniqueId": null, "ddlType": "ALTER_TABLE", "record_primary_key": null, "source_identity": null, "record_primary_value": null, "table_name": "all_mysql_type_test" }, "recordType": "DDL" } ``` ### Canal JSON 消息格式 数据同步至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 Canal 使用如下 JSON 消息格式。 ```JSON { "database": "STRING", // 使用 SQL 语句进行变更的数据库的名称。如果是 OceanBase 数据库，仅存在数据库名称，无需租户名称。 "sqlType": { "col1": "INTEGER", // 变更列类型，数字参考 java.sql.Types }, "data": [ // 变更后数据键值对，目前只会存在一条消息 { "col1": "val1" } ], "pkNames": [ // 主键列名 "col1" ], "old": [ // 仅 UPDATE 消息存在。表示 UPDATE 语句变更的列，即变更前的列值 { "col1": "val1" } ], "mysqlType": { // 列类型描述 "col": "STRING" }, "type": "STRING", // 变更类型 "table": "STRING", // 使用 SQL 语句进行变更的表的名称 "es": "LONG", // 变更时间，毫秒级时间戳 "isDdl": "BOOLEAN", // 是否是 DDL "ts": "LONG", // 写入目的端时间戳 "sql": "STRING" // 当前是空 } ``` 数据示例如下： - `INSERT`（插入）数据的示例 ```JSON { "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":1.2222, "col4":9.999999999999999093266253, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.9999999999999990, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "col1", "col2" ], "old":null, "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"INSERT", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618323429026, "sql":"" } ``` - `UPDATE`（更新）数据的示例 ```JSON { "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world 2020", "col3":1.2222, "col4":9.999999999999999093266253372484, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.9999999999999990932662, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "col1", "col2" ], "old":[ { "string":"hello world" } ], "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"UPDATE", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618364572908, "sql":"" } ``` - `DELETE`（删除）数据的示例 ```JSON { "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":1.2222, "col4":9.99999999999999909326625, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.99999999999999909326625, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "int8", "int16" ], "old":null, "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"DELETE", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618364660278, "sql":"" } ``` ```JSON { "database": "connector_test", "sqlType": null, "data": null, "pkNames": null, "old": null, "mysqlType": null, "type": "ALTER", "table": "all_mysql_type_test", "es": 1671177209000, "isDdl": true, "ts": 1671177291475, "sql": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" } ``` ### DataWorks JSON 消息格式 数据同步至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 DataWorks 使用如下 JSON 消息格式。 ```JSON { "version":"2.0", //协议版本，目前仅支持 DataWorks 2.0 版本 "schema": { //变更的元数据信息，仅指定列名与列类型信息 "source": {//变更来源信息 "dbType": "mysql", //数据源类型 "dbVersion": "5.7.35", //数据库版本 "dbName": "myDatabase", //数据库名称 "schema": "mySchema", //Schema 名称，存在 Schema 的系统必填 "table": "tableName" //表名 }, "column": [//变更的数据列信息，更新目标表记录内容 { "name": "id", "type": "bigint" }, { "name": "name", "type": "varchar(20)" }, { "name": "mydata", "type": "binary" }, { "name": "ts", "type": "datetime" } ], "pk": [//有主键或唯⼀键必填，否则可以不填 "pkName1", "pkName2" ] }, "payload": { "before": { "data":{ "id": 111, "name":"scooter", "mydata": "[base64 string]", //如果是二进制类型，需要进行 Base64 编码 "ts": 1590315269000.123456789 //时间戳，其整数部分 13 位，小数部分 9 位 } }, "after": { "data":{ "id": 222, "name":"donald", "mydata": "[base64 string]", "ts": 1590315269000 } }, "op":"INSERT/UPDATE/DELETE/HEARTBEAT/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/...",//大小写敏感 "timestamp": { "eventTime": 1620457659000 // 变更在源端库发生时间，毫秒精度的 13 位时间戳 }, "ddl": { "text": "ADD COLUMN ..." }, "scn": "⾃增 ID" }, "extend": { //extend 扩展字段，用于后续扩展需求。如果没有，可以不填 "load_fm":"CIBS" //记录来源系统。例如，"CIBS" } } ``` 同步任务心跳消息： ```JSON { "version": "2.0", //协议版本 "payload": { "timestamp": { "eventTime": 1620457659000 //⼼跳包时间 }, "op": "HEARTBEAT" //标识是⼼跳包 } } ``` 数据示例如下： ```JSON { "version":"2.0", "schema":{ "source":{ "dbType":"ob_mysql", "dbVersion":null, "dbName":"db", "schema":null, "table":"tab" }, "column":[ { "name":"int8", "type":"TINYINT" }, { "name":"int16", "type":"SMALLINT" }, { "name":"int32", "type":"INT" }, { "name":"int64", "type":"INT64" }, { "name":"float32", "type":"FLOAT" }, { "name":"float64", "type":"DOUBLE" }, { "name":"bigInt", "type":"BIGINT" }, { "name":"boolean", "type":"BOOLEAN" }, { "name":"string", "type":"VARCHAR" }, { "name":"bytes", "type":"BLOB" }, { "name":"decimal", "type":"DECIMAL" }, { "name":"localDate", "type":"DATE" }, { "name":"localTime", "type":"TIME" }, { "name":"localDateTime", "type":"DATETIME" }, { "name":"timestamp", "type":"TIMESTAMP" }, { "name":"zonedDateTime", "type":"ZONED_DATETIME" }, { "name":"intervalDayToSecond", "type":"INTERVAL_DAY_TO_SECOND" }, { "name":"intervalYearToMonth", "type":"INTERVAL_YEAR_TO_MONTH" } ], "pk":[ "pkName1", "pkName12" ] }, "payload":{ "before":null, "after":{ "data":{ "col1":3, "col2":129, "col3":2147483646, "col4":9223372036854775806, "col5":1.2222, "col6":0.0000000000000008125, "col7":10223372036854775806, "col8":1, "col9":"hello world", "col10":"aGVsbG8gd29ybGQ=", "col11":0.000000000000125, "col12":"2020-11-25", "col13":"00:01:02", "col14":"2020-11-25 00:01:02", "col15":"1606233662.012345", "col16":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col17":"INTERVAL '3' DAY", "col18":"INTERVAL '4' YEAR" } }, "op":"INSERT", "timestamp":{ "eventTime":1647581000000, "systemTime":1647581000795, "checkpointTime":1647581000 }, "ddl":null, "scn":"null" }, "extend":{ "load_fm": "test" } } ```JSON { "version":"2.0", "schema":{ "source":{ "dbType":"ob_mysql", "dbVersion":null, "dbName":"db", "schema":null, "table":"tab" }, "column":[ { "name":"int8", "type":"TINYINT" }, { "name":"int16", "type":"SMALLINT" }, { "name":"int32", "type":"INT" }, { "name":"int64", "type":"INT64" }, { "name":"float32", "type":"FLOAT" }, { "name":"float64", "type":"DOUBLE" }, { "name":"bigInt", "type":"BIGINT" }, { "name":"boolean", "type":"BOOLEAN" }, { "name":"string", "type":"VARCHAR" }, { "name":"bytes", "type":"BLOB" }, { "name":"decimal", "type":"DECIMAL" }, { "name":"localDate", "type":"DATE" }, { "name":"localTime", "type":"TIME" }, { "name":"localDateTime", "type":"DATETIME" }, { "name":"timestamp", "type":"TIMESTAMP" }, { "name":"zonedDateTime", "type":"ZONED_DATETIME" }, { "name":"intervalDayToSecond", "type":"INTERVAL_DAY_TO_SECOND" }, { "name":"intervalYearToMonth", "type":"INTERVAL_YEAR_TO_MONTH" } ], "pk":[ "pkName1", "pkName2" ] }, "payload":{ "before":{ "data":{ "col1":3, "col2":129, "col3":2147483646, "col4":9223372036854775806, "col5":1.2222, "col6":0.000000000000000125, "col7":10223372036854775806, "col8":1, "col9":"hello world", "col10":"aGVsbG8gd29ybGQ=", "col11":0.0000000000125, "col12":"2020-11-25", "col13":"00:01:02", "col14":"2020-11-25 00:01:02", "col15":"1606233662.012345", "col16":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col17":"INTERVAL '3' DAY", "col18":"INTERVAL '4' YEAR" } }, "after":{ "data":{ "col1":3, "col2":129, "col3":2147483646, "col4":9223372036854775806, "col5":1.2222, "col6":0.00000000008125, "col7":10223372036854775806, "col8":1, "col9":"hello world 2020", "col10":"aGVsbG8gd29ybGQ=", "col11":0.000000000125, "col12":"2020-11-25", "col13":"00:01:02", "col14":"2020-11-25 00:01:02", "col15":"1606233662.012345", "col16":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col17":"INTERVAL '3' DAY", "col18":"INTERVAL '4' YEAR" } }, "op":"UPDATE", "timestamp":{ "eventTime":1647581038000, "systemTime":1647581038674, "checkpointTime":1647581038 }, "ddl":null, "scn":"null" }, "extend":{ "load_fm": "test" } } ```JSON { "version":"2.0", "schema":{ "source":{ "dbType":"ob_mysql", "dbVersion":null, "dbName":"db", "schema":null, "table":"tab" }, "column":[ { "name":"int8", "type":"TINYINT" }, { "name":"int16", "type":"SMALLINT" }, { "name":"int32", "type":"INT" }, { "name":"int64", "type":"INT64" }, { "name":"float32", "type":"FLOAT" }, { "name":"float64", "type":"DOUBLE" }, { "name":"bigInt", "type":"BIGINT" }, { "name":"boolean", "type":"BOOLEAN" }, { "name":"string", "type":"VARCHAR" }, { "name":"bytes", "type":"BLOB" }, { "name":"decimal", "type":"DECIMAL" }, { "name":"localDate", "type":"DATE" }, { "name":"localTime", "type":"TIME" }, { "name":"localDateTime", "type":"DATETIME" }, { "name":"timestamp", "type":"TIMESTAMP" }, { "name":"zonedDateTime", "type":"ZONED_DATETIME" }, { "name":"intervalDayToSecond", "type":"INTERVAL_DAY_TO_SECOND" }, { "name":"intervalYearToMonth", "type":"INTERVAL_YEAR_TO_MONTH" } ], "pk":[ "pkName1", "pkName2" ] }, "payload":{ "before":{ "data":{ "col1":3, "col2":129, "col3":2147483646, "col4":9223372036854775806, "col5":1.2222, "col6":0.0000000000125, "col7":10223372036854775806, "col8":1, "col9":"hello world", "col10":"aGVsbG8gd29ybGQ=", "col11":0.0000000000125, "col12":"2020-11-25", "col13":"00:01:02", "col14":"2020-11-25 00:01:02", "col15":"1606233662.012345", "col16":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col17":"INTERVAL '3' DAY", "col18":"INTERVAL '4' YEAR" } }, "after":null, "op":"DELETE", "timestamp":{ "eventTime":1647581072000, "systemTime":1647581072976, "checkpointTime":1647581072 }, "ddl":null, "scn":"null" }, "extend":{ "load_fm": "test" } } ``` ```JSON { "version": "2.0", "schema": { "source": { "dbType": "ob_mysql", "dbVersion": null, "dbName": "connector_test", "schema": null, "table": "all_mysql_type_test" }, "column": null, "pk": null }, "payload": { "before": null, "after": null, "op": "ALTER", "timestamp": { "eventTime": 1671177209000, "systemTime": 1671177291485, "checkpointTime": 1671177200 }, "ddl": { "text": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" }, "scn": "null" }, "extend": {} } ``` ### SharePlex JSON 消息格式 数据同步至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 SharePlex 使用如下 JSON 消息格式。 ```JSON { "data": { // 变更数据键值对，如果是 INSERT / DELETE 是全量值，如果是 UPDATE 只有变更值 "col1": "val1" }, "meta": { "time": "YYYY-MM-DDTHH:mm:ss", // 变更时间 "op": "", // 变更类型，包括 ins/upd/del/ddl "posttime": "YYYY-MM-DDTHH:mm:ss", // 写入目标端的时间 "idx": "STRING", //消息在事务中的索引/索引的消息数量。该参数已废弃。 "size": "NUMBER", //事务内消息数量。该参数已废弃。 "seq": "STRING", // 排序序号，需要配合源端打开 transactionEnabled 才能存在 "table": "STRING", // SQL 变更库表名 {database}.{table} "rowid": "STRING", // {变更库表名}-{主键值使用\u0001} 分割 "trans": "STRING", // 事务 ID "scn": "STRING" // 该字段只有在增量场景下 source.json 配置中包含 sequenceEnabled=true 才存在，默认是 true。用于排序，生成规则是一个同步进程中，时间戳 + 不超过五位序号递增。 }, "key": { // 仅 UPDATE 存在，表示变更前的值 }, "sql": { "ddl": "" } // 仅 DDL 存在。如果 op = ddl，则会写入 DDL 语句。 } ``` 数据示例如下： ```JSON { "data":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.999999999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999999308, "col17":10223372036854775806, "col18":"1606233662.012345" }, "meta":{ "posttime":"2020-12-07T13:22:00", "op":"ins", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3129", "trans":"shareplex_transaction_id", "scn":"123456789" } } ```JSON { "data":{ "string":"hello world 2020" }, "meta":{ "posttime":"2020-12-07T13:59:09", "op":"upd", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3\u0001129", "trans":"shareplex_transaction_id", "scn":"123456789" }, "key":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.9999999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999999308, "col17":10223372036854775806, "col18":"1606233662.012345" } } ```JSON { "data":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.9999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999308, "col17":10223372036854775806, "col18":"1606233662.012345" }, "meta":{ "posttime":"2020-12-07T13:34:10", "op":"del", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3\u0001129", "trans":"shareplex_transaction_id", "scn":"123456789" } } ``` ```SQL { "data": {}, "meta": { "posttime": "2022-12-16T15:54:51", "op": "ddl", "size": 0, "time": "2022-12-16T15:53:29", "idx": "0/0", "seq": 0, "table": "connector_test.all_mysql_type_test", "rowid": "connector_test.all_mysql_type_test-", "trans": null, "scn": "null" }, "sql": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" comment 'test'" } } ``` ### DefaultExtendColumnType JSON 消息格式 数据同步至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 DefaultExtendColumnType 使用如下 JSON 消息格式。 DefaultExtendColumnType JSON 消息格式会在 `DEFAULT` 的基础上，在镜像内增加一个字段 `__light_type`，用于表示字段的数据类型。 ```JSON { "prevStruct": { // 变更前镜像 }, "postStruct": { // 变更后镜像 "__light_type": { "col": { // 字段的名称 "schemaType": "type" // 值的类型 } } }, "allMetaData": { } } ``` 数据示例如下： ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "INSERT", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.99999999999999909326625337248, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049338, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long":{ "schemaType":"TIMESTAMP" } } } } ``` ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": null, "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": null }, "prevStruct": null, "recordType": "ROW", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long":{ "schemaType":"TIMESTAMP" } } } } ```JSON { "allMetaData": { "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:1103909342127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType":"OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type": { "int8": { "schemaType": "TINYINT" }, "int16": { "schemaType": "SMALLINT" }, "int32": { "schemaType": "INT" }, "int64": { "schemaType": "INT64" }, "bigInt": { "schemaType": "BIGINT" }, "float32": { "schemaType": "FLOAT" }, "float64": { "schemaType": "DOUBLE" }, "string": { "schemaType": "VARCHAR" }, "bytes": { "schemaType": "BLOB" }, "decimal": { "schemaType": "DECIMAL" }, "localDate": { "schemaType": "DATE" }, "localTime": { "schemaType": "TIME" }, "localDateTime": { "schemaType": "DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } }, "recordType": "UPDATE", "postStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type": { "int8": { "schemaType": "TINYINT" }, "int16": { "schemaType": "SMALLINT" }, "int32": { "schemaType": "INT" }, "int64": { "schemaType": "INT64" }, "bigInt": { "schemaType": "BIGINT" }, "float32": { "schemaType": "FLOAT" }, "float64": { "schemaType": "DOUBLE" }, "string": { "schemaType": "VARCHAR" }, "bytes": { "schemaType": "BLOB" }, "decimal": { "schemaType": "DECIMAL" }, "localDate": { "schemaType": "DATE" }, "localTime": { "schemaType": "TIME" }, "localDateTime": { "schemaType": "DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } } } ```JSON { "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:1103xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // 此处为 OceanBase 数据库 V4.x 及之后版本的示例。OceanBase 数据库 V4.x 之前版本的示例为 "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}"。 "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } }, "recordType":"DELETE", "postStruct":null } ``` ```JSON { "prevStruct": null, "postStruct": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'", "__light_type": { "ddl": { "schemaType": "VAR_STRING" } } }, "allMetaData": { "checkpoint": "1671177200", "dbType": "OB_MYSQL", "storeDataSequence": null, "db": "connector_test", "timestamp": "1671177209", "uniqueId": null, "ddlType": "ALTER_TABLE", "record_primary_key": null, "source_identity": null, "record_primary_value": null, "table_name": "all_mysql_type_test" }, "recordType": "DDL" } ``` ### Debezium JSON 消息格式 同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 Debezium 使用如下 JSON 消息格式，共包含两种，通常默认仅显示 `payload` 中的结构。 - 存在 `schema` 和 `payload` ```JSON { "schema": { //描述 payload 字段信息的结构体，默认没有该结构体 "type": "struct", //struct 表示该字段内部还有结构 "optional": false, //是否必须包含该字段 "fields": [ { "type": "int64", //字段的类型 "optional": false, //是否必须包含该字段 "field": "ts_ms" //字段的名称 } ] }, "payload": { "op": "c", //数据修改类型，包括 c（全量、插入）、u（更新）、d（删除）和 HEARTBEAT（心跳消息） "source": { "version": "", //OMS 的版本 "connector": "OB_MYSQL", //数据源的类型 "name": "OMS", //固定值 OMS "ts_ms": 0, //数据变更秒级时间戳，仅增量存在 "db": "test", //使用 SQL 语句进行变更的数据库的名称。如果是 OceanBase 数据库，仅存在数据库名称，无需租户名称 "table": "testTab", //使用 SQL 语句进行变更的表的名称 "pos": "553132@1668496109" //在 binlog 文件中的位置 [binlog 文件名]@[binlog 文件名 offset] }, "before": { //变更前镜像 "column": "value" //键值对，包含全量键值 }, "after": { //变更后镜像 "column": "value" // 键值对，包含全量键值 }, "ts_ms": 1668497367188 //数据处理时间戳 } } ``` - 仅存在 `payload` ```JSON { "payload": { "op": "c", //数据修改类型，包括 c（全量、插入）、u（更新）、d（删除）和 HEARTBEAT（心跳消息） "source": { "version": "", //OMS 的版本 "connector": "OB_MYSQL", //数据源的类型 "name": "OMS", //固定值 OMS "ts_ms": 0, //数据变更秒级时间戳，仅增量存在 "db": "test", //使用 SQL 语句进行变更的数据库的名称。如果是 OceanBase 数据库，仅存在数据库名称，无需租户名称 "table": "testTab", //使用 SQL 语句进行变更的表的名称 "pos": "553132@16684****" //在 binlog 文件中的位置 [binlog 文件名]@[binlog 文件名 offset] }, "before": { //变更前镜像 "column": "value" //键值对，包含全量键值 }, "after": { //变更后镜像 "column": "value" // 键值对，包含全量键值 }, "ts_ms": 1668497367188 //数据处理时间戳 } } ``` 数据示例如下： ```JSON { "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"source", "optional":false, "type":"struct", "fields":[ { "field":"version", "optional":false, "type":"string" }, { "field":"connector", "optional":false, "type":"string" }, { "field":"name", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" }, { "field":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"c", "source":{ "connector":"OB_MYSQL", "pos":"703223@166849****", "name":"OMS", "version":"", "ts_ms":1668491621000, "db":"test", "table":"table_name" }, "after":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":2, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495423594 } } ```JSON { "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"source", "optional":false, "type":"struct", "fields":[ { "field":"version", "optional":false, "type":"string" }, { "field":"connector", "optional":false, "type":"string" }, { "field":"name", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" }, { "field":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"u", "before":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"6900000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"436999@166849****", "name":"OMS", "version":"", "ts_ms":1668495861000, "db":"test", "table":"table_name" }, "after":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495906356 } } ```JSON { "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"source", "optional":false, "type":"struct", "fields":[ { "field":"version", "optional":false, "type":"string" }, { "field":"connector", "optional":false, "type":"string" }, { "field":"name", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" }, { "field":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"d", "before":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"553132@1668****", "name":"OMS", "version":"", "ts_ms":1668496109000, "db":"test", "table":"table_name" }, "ts_ms":1668496119717 } } ``` ### DebeziumFlatten JSON 消息格式 同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 DebeziumFlatten 的 JSON 消息格式如下所示，和序列化方式 Debezium 相比，不再填充 `schema` 和 `payload`。 ```JSON { "op": "c", //数据修改类型，包括 c（全量、插入）、u（更新）、d（删除）和 HEARTBEAT（心跳消息） "source": { "version": "", //OMS 的版本 "connector": "OB_MYSQL", //数据源的类型 "name": "OMS", //固定值 OMS "ts_ms": 0, //数据变更秒级时间戳，仅增量存在 "db": "test", //使用 SQL 语句进行变更的数据库的名称。如果是 OceanBase 数据库，仅存在数据库名称，无需租户名称 "table": "testTab", //使用 SQL 语句进行变更的表的名称 "pos": "553132@16684****" //在 binlog 文件中的位置 [binlog 文件名]@[binlog 文件名 offset] }, "before": { //变更前镜像 "column": "value" //键值对，包含全量键值 }, "after": { //变更后镜像 "column": "value" // 键值对，包含全量键值 }, "ts_ms": 1668497367188 //数据处理时间戳 } ``` 数据示例如下： ```JSON { "op":"c", "source":{ "connector":"OB_MYSQL", "pos":"703223@166849****", "name":"OMS", "version":"", "ts_ms":1668491621000, "db":"test", "table":"table_name" }, "after":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":2, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495423594 } ```JSON { "op":"u", "before":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"690000000000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"436999@166849****", "name":"OMS", "version":"", "ts_ms":1668495861000, "db":"test", "table":"table_name" }, "after":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495906356 } ```JSON { "op":"d", "before":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"553132@1668****", "name":"OMS", "version":"", "ts_ms":1668496109000, "db":"test", "table":"table_name" }, "ts_ms":1668496119717 } ``` ### DebeziumSmt JSON 消息格式 DebeziumSmt 是 Debezium 提供的一种配置方式，使用事件扁平化单消息转换（Single Message Transform，SMT）对单条信息进行转换和处理。同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时，序列化方式 DebeziumSmt 的 JSON 消息格式仅显示 `after` 中的 `key:value`。 例如，使用序列化方式 Debezium 更新数据： ```JSON { "op": "u", "source": { "connector": "OB_MYSQL", "name": "OMS" }, "ts_ms": 1668496119717, "before": { "field1": "before_value1", "field2": "before_value2" }, "after": { "field1": "after_value1", "field2": "after_value2" } } ``` SMT 对上述示例的消息进行处理后，简化了消息格式。即使用序列化方式 DebeziumSmt，JSON 消息格式如下所示。 ```JSON { "field1": "after_value1", "field2": "after_value2" } ``` 数据示例如下： ```JSON { "field1": "after_value1", "field2": "after_value2", "__deleted": "false" } ```JSON { "field1": "after_value1", "field2": "after_value2", "__deleted": "true" } ``` ### Avro JSON 消息格式 同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka 时，序列化方式 **Avro** 使用如下 JSON 消息格式。 - 全量迁移 ```JSON { "version": 1, "id": 0, "sourceTimestamp": 1702371565, // 时间戳安全位点。 "sourcePosition": "", // 全量迁移无 position 等信息。 "safeSourcePosition": "", "sourceTxid": "", "source": { "sourceType": "MySQL", // 固定值 MySQL。 "version": "OBMySQL" // 固定值 OBMySQL。 }, "operation": "INIT", // 全量类型为 INIT。 "objectName": "test***", "processTimestamps": [ 1702371565238 ], // 只有投递时间。 "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" // 只有主键类型。 }, "fields": [ { "name": "id", "dataTypeNumber": 246 }, // 每个列的类型。 { "name": "bid", "dataTypeNumber": 3 }, { "name": "name", "dataTypeNumber": 15 }, { "name": "address", "dataTypeNumber": 254 } ], "beforeImages": null, // 全量迁移前镜像为空。 "afterImages": [ // 后镜像。INTEGER 类型的 precision 衡为 8，FLOAT 类型的 precision 衡为 8、scale 衡为 64。 { "value": "1", "precision": 1, "scale": 0 }, { "precision": 8, "value": "11" }, { "charset": "utf8mb4", "value": { "bytes": "yyy" } }, null ] } ``` - 增量同步 DML ```JSON { "version": 1, "id": 170236922143600000, "sourceTimestamp": 1702369092, "sourcePosition": "1702369080", // OceanBase 数据库 MySQL 兼容模式的 checkpoint。 "safeSourcePosition": "1702369080", // OceanBase 数据库 MySQL 兼容模式的 checkpoint。 "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "INSERT", "objectName": "test***", "processTimestamps": [1702369221480], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": null, // INSERT 前镜像为空。 "afterImages": [ {"precision": 8, "value": "2"}, {"precision": 8, "value": "12"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"} } ] } ```JSON { "version": 1, "id": 170236975822100001, "sourceTimestamp": 1702369757, "sourcePosition": "1702369756", "safeSourcePosition": "1702369756", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "UPDATE", "objectName": "test***", "processTimestamps": [1702369758237], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": [ // UPDATE 存在前镜像和后镜像。 {"precision": 8, "value": "3"}, {"precision": 8, "value": "22"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ], "afterImages": [ {"precision": 8, "value": "3"}, {"precision": 8, "value": "44"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ] } ```JSON { "version": 1, "id": 170236976527500000, "sourceTimestamp": 1702369764, "sourcePosition": "1702369763", "safeSourcePosition": "1702369763", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "DELETE", "objectName": "test***", "processTimestamps": [1702369765287], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": [ {"precision": 8, "value": "3"}, {"precision": 8, "value": "44"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ], "afterImages": null // DELETE 后镜像为空。 } ``` - 增量同步 DDL ```JSON { "version": 1, "id": 170236979372400000, "sourceTimestamp": 1702369793, "sourcePosition": "1702369792", "safeSourcePosition": "1702369792", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "DDL", "objectName": "test***", "processTimestamps": [1702369794543], "tags": {}, "fields": null, // 增量同步 DDL 无 fields 和 beforeImages。 "beforeImages": null, "afterImages": "alter table multi_db_multi_tbl add column address char(20) default null" // STRING 类型的 afterImages 为 DDL 语句。 } ``` ## 数据库传输到文本协议的格式说明 同步 OceanBase 数据库的数据至 Kafka、DataHub（BLOB 类型）和 RocketMQ 时 - 如果序列化方式为 **Default**、 **Canal**、**DataWorks**（支持 V2.0）、 **SharePlex** 或 **DefaultExtendColumnType**，OceanBase 数据库两种兼容模式对应的映射说明如下。 - OceanBase 数据库 MySQL 兼容模式 | **数据类型** | **映射类型** | **描述** | | --- | --- | --- | | TINYINT SMALLINT MEDIUMINT INT INTEGER YEAR BOOL BOOLEAN | Long | 64 位以下的整型。 正常数值，例如 1000，不使用科学计数法。 对于 BOOL/BOOLEAN，则 true = 1，false = 0。 | | DECIMAL NUMERIC | BigDecimal | 精确小数数值类型以及超过 64 位的整型。对于整型数值不会展示小数点及小数。对于存在小数的数值，会根据数据库传入的数据进行位数展示，不会去除末尾的 0，使用科学计数法。 | | FLOAT DOUBLE | Double | 浮点数 根据源端是 FLOAT 或 DOUBLE 类型决定有效位数。FLOAT 是 7 位有效位数，DOUBLE 是 16 位有效位数。 | | CHAR VARCHAR TINYTEXT TEXT MEDIUMTEXT LONGTEXT ENUM SET | String | 字符串。 | | TINYBLOB BLOB MEDIUMBLOB LONGBLOB BINARY VARBINARY BIT | Bytes | 字节数组，默认以 BASE64 编码展示。 **说明** 对于 BIT 定长的类型，增量接收到字节数组之后会将高位 0 去除，但是全量不会，所以看到的 BASE64 编码可能会不一致。但是实际结果一致，解码之后结果一致。 | | DATE | Date | 日期类型，格式为 `YYYY-MM-DD`。 如果是非法时间，会显示原有字符串。 | | TIME | Time | 时间类型，格式为 `HH:mm:ss[.nnnnnnnnn]`。 低于秒级的时间最多展示 9 位。如果是低于秒级的时间，会显示出所有非 0 的数值。如果是非法时间，会显示原有字符串。 | | DATETIME | DateTime | 日期时间类型，包括时区。格式为 `YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId]`。 低于秒级的时间最多展示 9 位。如果是低于秒级的时间，会显示出所有非 0 的数值。如果是非法时间，会显示原有字符串。 | | TIMESTAMP | Timestamp | 时间戳类型，格式为 `[秒级时间戳][.nnnnnnnnn]`。 低于秒级的时间最多展示 9 位。如果是低于秒级的时间，会显示出所有非 0 的数值。如果是非法时间，会以 0000-00-00 00:00:00 格式显示。 | - OceanBase 数据库 Oracle 兼容模式 | **数据类型** | **映射类型** | **描述** | | --- | --- | --- | | INTEGER | Long | 64 位以下的整型。 正常数值，例如 1000，不使用科学计数法。 | | NUMBER FLOAT | BigDecimal | 精确小数数值类型以及超过 64 位的整型。 | | BINARY_FLOAT BINARY_DOUBLE | Double | 浮点数 根据源端是 FLOAT 或 DOUBLE 类型决定有效位数。FLOAT 是 7 位有效位数，DOUBLE 是 16 位有效位数。 | | VARCHAR2 NVARCHAR2 INTERVAL YEAR TO MOTH INTERVAL DAY TO SECOND CLOB NCLOB ROWID UROWID | String | 字符串 | | BLOB BFILE RAW | Bytes | 字节数组 默认以 BASE64 编码展示。 | | DATE TIMESTAMP TIMESTAMP WITH TIME ZONE TIMESTAMP WITH LOCAL TIME ZONE | DateTime | 日期时间类型，包括时区。格式为 `YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId]`。 低于秒级的时间最多展示 9 位。如果是低于秒级的时间，会显示出所有非 0 的数值。如果是非法时间，会显示原有字符串。 | - 如果序列化方式为 **Debezium**，OceanBase 数据库 MySQL 兼容模式对应的映射说明如下。 #### 注意 同步 OceanBase 数据库 Oracle 兼容模式的数据至 Kafka、DataHub（Blob 类型）和 RocketMQ 时，不支持选择序列化方式 **Debezium**。 | 数据类型 | 映射类型 | 描述 | | --- | --- | --- | | BOOLEAN BOOL | BOOLEAN | 取值包括 true 和 false。 | | TINYINT SMALLINT MEDIUMINT INT/INTEGER BIGINT YEAR | LONG | -2^63^ ～ 2^63^范围的整型。 | | BIGINT | STRING | 使用字符串完整展示数据。 | | FLOAT DOUBLE | DOUBLE | 浮点数。 | | DECIMAL NUMERIC | STRING | 使用字符串完整展示数据。对于存在小数的数值，会根据数据库传入的数据进行位数展示，不会去除末尾的 0，使用科学计数法。 | | BIT BINARY VARBINARY TINYBLOB BLOB MEDIUMBLOB LONGBLOB | BYTES | 字节数组，base16 编码。 | | CHAR VARCHAR TINYTEXT TEXT MEDIUMTEXT LONGTEXT ENUM SET | STRING | 字符串。 | | TIMESTAMP | STRING | 格式为 YYYY-MM-DDTHH:mm:ss[.nnnnnnnnn]Z，时区为 0 时区。 | | DATE | LONG | 表示自 1970-01-01 以来的天数。 | | TIME | LONG | 表示自 00:00:00 以来的时间值（以微秒为单位），不包括时区信息。 | | DATETIME | LONG | 表示自 1970-01-01 00:00:00 以来的毫秒数，不包括时区信息。 | - 如果序列化方式为 **Avro**，OceanBase 数据库 MySQL 兼容模式对应的映射说明如下。 #### 注意 仅同步 OceanBase 数据库 MySQL 兼容模式的数据至 Kafka 时，支持选择序列化方式 **Avro**。 | 类型名称 | 映射类型 | | --- | --- | | TINYINT BOOLEAN SMALLINT MEDIUMINT INT BIGINT BIT | INTEGER | | FLOAT DOUBLE | FLOAT | | DECIMAL NUMERIC | DECIMAL | | VARCHAR CHAR TINYTEXT MEDIUMTEXT LONGTEXT TEXT | CHARACTER | | BINARY VARBINARY TINYBLOB MEDIUMBLOB LONGBLOB BLOB | BinaryObject | | TIMESTAMP | TimestampObject **说明** 对于 TIMESTAMP 类型，全量和增量均会转换至时间戳，非法时间为 `-9223372022400L`。 除非法时间外，您可以使用 Java 的 `Instant.ofEpochSecond(ts, nanos)` 方法获取正确的墙上时间。 | | DATE TIME DATETIME YEAR | DATETIME | | JSON ENUM SET | TextObject | | GEOMETRY | TextGeometry **说明** 目前数据传输使用 EWKT 格式透传，所以映射为 TextGeometry 类型。 | 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线