Este tema describe los formatos de datos que se utilizan en los métodos de serialización y cuando se transmiten datos desde OceanBase Database hacia un sistema de colas de mensajes.
Formatos de datos en métodos de serialización
Al usar el servicio de transmisión de datos para sincronizar información desde el origen hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, puede emplear un método de serialización para controlar el formato de los datos sincronizados. Los métodos de serialización compatibles son Default, Canal, DataWorks (V2.0), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten, DebeziumSmt y Avro.
Nota
Actualmente, solo los tenants de MySQL de OceanBase Database admiten los métodos de serialización Debezium, DebeziumFlatten y DebeziumSmt.
Por ahora, el método de serialización Avro solo es compatible cuando se sincronizan datos desde un tenant de MySQL de OceanBase Database hacia una instancia de Kafka.
Formato de mensaje JSON predeterminado
Cuando sincroniza datos hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización Default:
{
"prevStruct": { // The image before modification.
"col1": "val1" // The key-value pair that contains the full key value.
},
"postStruct": { // The image after modification.
"col1": "val1" // The key-value pair that contains the full key value.
},
"allMetaData"{
"checkpoint": "STRING", // The current synchronization checkpoint, which specifies the target checkpoint (second-level timestamp) in the incremental synchronization phase and the primary key-value pair in the full synchronization phase.
"record_primary_key": "STRING", // The name of the primary key column. If the primary key has multiple columns, separate the column names with \u0001.
"record_primary_value": "STRING", // The primary key value. If the primary key has multiple columns, separate the column names with \u0001.
"source_identity": "STRING", // The identifier of the source. The value is a subtopic in incremental migration and a meaningless sequence number in full migration.
"dbType": "STRING", // The database type. Valid values: MYSQL, ORACLE, OCEANBASE, OB_IN_ORACLE_MODE, DB2, OB_MYSQL, OB_ORACLE, and DB2_LUW. Note that OCEANBASE, OB_IN_ORACLE_MODE, and DB2 are used by previous versions of the data transmission service and are compatible with this version of the data transmission service.
"storeDataSequence": "LONG", // This field is available only when the value of sequenceEnabled is set to true in the source.json configuration file in incremental synchronization scenarios. Default value: true. This field is used for sorting. A sequence is generated during the synchronization process in the timestamp + sequence number format. The sequence number increments and does not exceed five digits. {Timestamp}{Incrementing sequence number}.
"table_name": "STRING", // The name of the table that is modified by using an SQL statement.
"db": "STRING", // The name of the database that is modified by using an SQL statement. For an OceanBase database, the database name includes the tenant name. The format is {tenant}.{database}.
"timestamp": "STRING", // The timestamp in seconds for data modification, which is available only for incremental synchronization.
"uniqueId": "STRING", // The sequence number as the identifier that is passed from the store in incremental migration.
"ddlType": "STRING", // The DDL operation type.
},
"recordType": "INSERT/UPDATE/DELETE/HEARTBEAT/DDL" // The modification type.
}Si los registros DDL contienen únicamente claves con el nombre de columna "ddl", los valores corresponden a sentencias DDL.
-
Imagen anterior y posterior:
prevStruct: imagen anterior de los datos incrementales; corresponde a los datos antes de ejecutar una sentencia SQL.postStruct: imagen posterior de los datos incrementales; corresponde a los datos después de ejecutar una sentencia SQL.
DELETEcontiene solo el campoprevStruct.INSERTyDDLcontienen únicamente el campopostStruct.UPDATEincluye tantoprevStructcomopostStruct.HEARTBEAT(que indica un mensaje de latido periódico) no contienepostStructnipostStruct.
A continuación se muestran algunos ejemplos:
-
Ejemplo de inserción de datos (INSERT)
{ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table_name", "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" } } -
Ejemplo de actualización de datos (UPDATE)
{ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table_name", "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" } } -
Ejemplo de eliminación de datos (DELETE)
{ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table_name", "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 } -
Ejemplo de una operación DDL
alter table connector_test.all_mysql_type_test add column c90 varchar(30) default "test" comment 'test';{ "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" }
Formato de mensaje JSON de Canal
Cuando sincroniza datos hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización Canal:
{
"database": "STRING", // The name of the database that is modified by using an SQL statement. For an OceanBase database, you can specify only the database name and omit the tenant name.
"sqlType": {
"col1": "INTEGER" // The data type of the column. For more information, see java.sql.Types.
},
"data": [ // The modified data key-value pair. This field exists in only one message.
{
"col1": "val1"
}
],
"pkNames": [ // The primary key column name.
"col1"
],
"old": [ // This field exists only in update messages and indicates the column modified by using an UPDATE statement. It indicates the column value before the modification.
{
"col1": "val1"
}
],
"mysqlType": { // The description of the column type.
"col": "STRING"
},
"type": "STRING", // The change type.
"table": "STRING", // The name of the table that is modified by using an SQL statement.
"es": "LONG", // The time of modification, which is a timestamp in milliseconds.
"isDdl": "BOOLEAN", // Indicates whether the statement is a DDL statement.
"ts": "LONG", // The timestamp at which data is written to the target.
"sql": "STRING" // It is empty.
}A continuación se muestran algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col5": 129, "col6": "00:01:02", "col7": 2147483646, "col8": 9223372036854775806, "col9": "blob_data", "col10": 3, "col11": "2020-11-25", "col12": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col13": 10223372036854775806, "col14": "2024-09-26 16:22:43" } ], "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": "" } -
Ejemplo de actualización de datos (
UPDATE){ "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col5": 129, "col6": "00:01:02", "col7": 2147483646, "col8": 9223372036854775806, "col9": "blob_data", "col10": 3, "col11": "2020-11-25", "col12": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col13": 10223372036854775806, "col14": "2024-09-26 16:22:43" } ], "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": "" } -
Ejemplo de eliminación de datos (
DELETE){ "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col5": 129, "col6": "00:01:02", "col7": 2147483646, "col8": 9223372036854775806, "col9": "blob_data", "col10": 3, "col11": "2020-11-25", "col12": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col13": 10223372036854775806, "col14": "2024-09-26 16:22:43" } ], "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": "" } -
Ejemplo de una operación DDL
alter table connector_test.all_mysql_type_test add column c90 varchar(30) default "test" comment 'test';{ "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'" }
Formato de mensaje JSON de DataWorks
Cuando sincroniza datos hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización DataWorks:
{
"version": "2.0", // The protocol version. Only DataWorks 2.0 is supported.
"schema": { // The modified metadata information, with only the column name and column type specified.
"source": { // The source of the modification.
"dbType": "mysql", // The source type.
"dbVersion": "5.7.35", // The database version.
"dbName": "myDatabase", // The database name.
"schema": "mySchema", // The schema name. This field is required in a system with schemas.
"table": "tableName" // The table name.
},
"column": [ // The modified data column. This field indicates the updated record content in the target table.
{
"name": "id",
"type": "bigint"
},
{
"name": "name",
"type": "varchar(20)"
},
{
"name": "mydata",
"type": "binary"
},
{
"name": "ts",
"type": "datetime"
}
],
"pk": [ // This field is required if the table has a primary key or unique key. Otherwise, it is optional.
"pkName1",
"pkName2"
]
},
"payload": {
"before": {
"data": {
"id": 111,
"name": "scooter",
"mydata": "[base64 string]", // Base64 encoding is required for the binary type.
"ts": 1590315269000.123456789 // The timestamp with an integer part of 13 digits and a fractional part of 9 digits.
}
},
"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/...", // The operation type. The value is in uppercase.
"timestamp": {
"eventTime": 1620457659000 // The modification time on the source database, which is a 13-digit timestamp in milliseconds.
},
"ddl": {
"text": "ADD COLUMN ..."
},
"scn": "The system change number (SCN) that increments."
},
"extend": { // The extend field that can be used for meeting extension requirements in the future. This field can be left unspecified if no extension is available.
"load_fm": "CIBS" // The source system. for example, "CIBS".
}
}Mensaje de latido de una tarea de sincronización:
{
"version": "2.0", // The protocol version.
"payload": {
"timestamp": {
"eventTime": 1620457659000 // The timestamp of the heartbeat packet.
},
"op": "HEARTBEAT" // Indicates a heartbeat packet.
}
}
A continuación se muestran algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "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.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "col7": 10223372036854775806, "col8": 1, "col9": "hello world", "col10": "aGVsbG8gd29ybGQ=", "col11": 0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "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" } } -
Ejemplo de actualización de datos (
UPDATE){ "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.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "col7": 10223372036854775806, "col8": 1, "col9": "hello world", "col10": "aGVsbG8gd29ybGQ=", "col11": 0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "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.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "col7": 10223372036854775806, "col8": 1, "col9": "hello world 2020", "col10": "aGVsbG8gd29ybGQ=", "col11": 0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "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" } } -
Ejemplo de eliminación de datos (
DELETE){ "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.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "col7": 10223372036854775806, "col8": 1, "col9": "hello world", "col10": "aGVsbG8gd29ybGQ=", "col11": 0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000999999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125, "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" } } -
Ejemplo de una operación DDL
alter table connector_test.all_mysql_type_test add column c90 varchar(30) default "test" comment 'test';{ "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": {} }
Formato de mensaje JSON de SharePlex
Al sincronizar datos con una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización SharePlex:
{
"data": { // The modified data key-value pair. If an INSERT or a DELETE statement is executed, the value is the full value. If an UPDATE statement is executed, the value is the modified value.
"col1": "val1"
},
"meta": {
"time": "YYYY-MM-DDTHH:mm:ss", // The time of modification.
"op": "", // The change type. Valid values: ins, upd, del, and ddl.
"posttime": "YYYY-MM-DDTHH:mm:ss", // The time at which data is written to the target.
"idx": "STRING", // Index of the message in the transaction/Number of index messages. This field is deprecated.
"size": NUMBER, // The number of messages in the transaction. This field is deprecated.
"seq": "STRING", // The sorting sequence. This field is available only when transactionEnabled is set to true in the source database.
"table": "STRING", // The database and table name in the {database}.{table} format modified by using an SQL statement.
"rowid": "STRING", // {The modified database and table name}-{The primary key values separated with \u0001}.
"trans": "STRING", // The transaction ID.
"scn": "STRING" // This field is available only when the value of sequenceEnabled is set to true in the source.json configuration file in incremental synchronization scenarios. Default value: true. This field is used for sorting. A sequence is generated during the synchronization process in the timestamp + sequence number format. The sequence number increments and does not exceed five digits.
},
"key": { // The key value before modification, which is available only for modifications by using an UPDATE statement.
}
}
A continuación se presentan algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "data": { "col1": "2020-11-25 00:01:02", "col2": "hello world", "col3": "INTERVAL '3' DAY", "col4": 1.2222, "col5": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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" } } -
Ejemplo de actualización de datos (
UPDATE){ "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col17": 10223372036854775806, "col18": "1606233662.012345" } } -
Ejemplo de eliminación de datos (
DELETE){ "data": { "col1": "2020-11-25 00:01:02", "col2": "hello world", "col3": "INTERVAL '3' DAY", "col4": 1.2222, "col5": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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" } } -
Ejemplo de una operación DDL
alter table connector_test.all_mysql_type_test add column c90 varchar(30) default "test" comment 'test';{ "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'" } }
Formato de mensaje JSON de DefaultExtendColumnType
Al sincronizar datos con una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización DefaultExtendColumnType:
Los mensajes JSON de DefaultExtendColumnType se generan en formato Default e incluyen un campo __light_type. Este campo especifica el tipo de dato de los campos.
{
"prevStruct": { // The image before modification.
},
"postStruct": { // The image after modification.
"__light_type": {
"col": { // The field name.
"schemaType": "type" // The value type.
}
}
},
"allMetaData": {
}
}
A continuación se presentan algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "INSERT", "postStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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" } } } } -
Ejemplo de actualización de datos (
UPDATE){ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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" } } } } -
Ejemplo de eliminación de datos (
DELETE){ "allMetaData": { "checkpoint": null, "record_primary_key": "id1\u0001id2", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049331647881341002217023668530611028595157578301758491822824378438792553200763769833775473829862512856683413461939989729065436937279228852476622948659167943435544622149348072943613294167216662821737555414480159115639791276054897201420389770580351533960771506199055664889770260291710977826725024401716523031627390652604144008597950935492433262042405635563993262949691698930975461134804791235994697938405200089317860731205010159117711704697471514344499487123311264707354172378099538737850219826145102366279591379660471881259976727356521602405329789906247763521525981391443887618575275588619928089116905061711975308467857756405810961619074331866883961080943542712559830853980002984826569445431232452392578125E-308, "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 } -
Ejemplo de una operación DDL
alter table connector_test.all_mysql_type_test add column c90 varchar(30) default "test" comment 'test';{ "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" }
Formato de mensaje JSON de Debezium
Al sincronizar datos desde un inquilino MySQL de OceanBase Database hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utilizan los siguientes formatos de mensaje JSON para el método de serialización Debezium. Por lo general, solo la estructura del campo payload se muestra de forma predeterminada.
-
Formato donde se especifican los campos
schemaypayload{ "schema": { // The struct that describes the payload field. By default, this struct is not specified. "type": "struct", // struct indicates that this field contains other structs. "optional": false, // Specifies whether this field is required. "fields": [ { "type": "int64", // The field type. "optional": false, // Specifies whether this field is required. "field": "ts_ms" // The field name. } ] }, "payload": { "op": "c", // The type of data modification. Valid values: c, u, d, and HEARTBEAT. c indicates full insert. u indicates update. d indicates delete. HEARTBEAT indicates a heartbeat message. "source": { "version": "", // The OMS version. "connector": "OB_MYSQL", // The type of the data source. "name": "OMS", // A fixed value, which is OMS. "ts_ms": 0, // The timestamp in seconds for data modification, which is available only for incremental synchronization. "db": "test", // The name of the database that is modified by using an SQL statement. For an OceanBase database, you can specify only the database name and omit the tenant name. "table": "testTab", // The name of the table that is modified by using an SQL statement. "pos": "553132@1668496109" // The log location in the binlog file, in the format of [binlog file name]@[binlog file name offset]. }, "before": { // The image before modification. "column": "value" // The key-value pair that contains the full key value. }, "after": { // The image after modification. "column": "value" // The key-value pair that contains the full key value. }, "ts_ms": 1668497367188 // The timestamp of data processing. } } -
Formato donde solo se especifica el campo
payload{ "payload": { "op": "c", // The type of data modification. Valid values: c, u, d, and HEARTBEAT. c indicates full insert. u indicates update. d indicates delete. HEARTBEAT indicates a heartbeat message. "source": { "version": "", // The OMS version. "connector": "OB_MYSQL", // The type of the data source. "name": "OMS", // A fixed value, which is OMS. "ts_ms": 0, // The timestamp in seconds for data modification, which is available only for incremental synchronization. "db": "test", // The name of the database that is modified by using an SQL statement. For an OceanBase database, you can specify only the database name and omit the tenant name. "table": "testTab", // The name of the table that is modified by using an SQL statement. "pos": "553132@16684****" // The log location in the binlog file, in the format of [binlog file name]@[binlog file name offset]. }, "before": { // The image before modification. "column": "value" // The key-value pair that contains the full key value. }, "after": { // The image after modification. "column": "value" // The key-value pair that contains the full key value. }, "ts_ms": 1668497367188 // The timestamp of data processing. } }
A continuación se presentan algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "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": "690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 } } -
Ejemplo de actualización de datos (
UPDATE){ "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":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 } } -
Ejemplo de eliminación de datos (
DELETE){ "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":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 } }
Formato de mensaje JSON de DebeziumFlatten
Al sincronizar datos desde un inquilino MySQL de OceanBase Database hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, se utiliza el siguiente formato de mensaje JSON para el método de serialización DebeziumFlatten. A diferencia del método de serialización Debezium, los campos schema y payload no se proporcionan en el método de serialización DebeziumFlatten.
{
"op": "c", // The type of data modification. Valid values: c, u, d, and HEARTBEAT. c indicates full insert. u indicates update. d indicates delete. HEARTBEAT indicates a heartbeat message.
"source": {
"version": "", // The OMS version.
"connector": "OB_MYSQL", // The type of the data source.
"name": "OMS", // A fixed value, which is OMS.
"ts_ms": 0, // The timestamp in seconds for data modification, which is available only for incremental synchronization.
"db": "test", // The name of the database that is modified by using an SQL statement. For an OceanBase database, you can specify only the database name and omit the tenant name.
"table": "testTab", // The name of the table that is modified by using an SQL statement.
"pos": "553132@16684****" // The log location in the binlog file, in the format of [binlog file name]@[binlog file name offset].
},
"before": { // The image before modification.
"column": "value" // The key-value pair that contains the full key value.
},
"after": { // The image after modification.
"column": "value" // The key-value pair that contains the full key value.
},
"ts_ms": 1668497367188 // The timestamp of data processing.
}A continuación se presentan algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "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":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 } -
Ejemplo de actualización de datos (
UPDATE){ "op":"u", "before":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 } -
Ejemplo de eliminación de datos (
DELETE){ "op":"d", "before":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"690000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", "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 }
Formato de mensaje JSON de DebeziumSmt
DebeziumSmt es un método de configuración proporcionado por el método de serialización Debezium. Este método transforma y procesa un mensaje individual mediante la transformación de mensaje único (SMT) de aplanamiento de eventos. Cuando sincroniza datos desde un inquilino MySQL de OceanBase Database hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, solo se muestran los pares key:value de la sección after del formato de mensaje JSON correspondiente al método de serialización DebeziumSmt.
En el siguiente ejemplo, se utiliza el método de serialización Debezium para actualizar datos:
{
"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"
}
}La SMT simplifica el formato del mensaje anterior. El formato de mensaje JSON simplificado con el método de serialización DebeziumSmt es el siguiente:
{
"field1": "after_value1",
"field2": "after_value2"
}A continuación se presentan algunos ejemplos:
-
Ejemplo de inserción de datos (
INSERT){ "field1": "after_value1", "field2": "after_value2", "__deleted": "false" } -
Ejemplo de actualización de datos (
UPDATE){ "field1": "after_value1", "field2": "after_value2", "__deleted": "false" } -
Ejemplo de eliminación de datos (
DELETE){ "field1": "after_value1", "field2": "after_value2", "__deleted": "true" }
Formato de mensaje JSON de Avro
Cuando sincroniza datos desde un inquilino MySQL de OceanBase Database hacia una instancia de Kafka, se utiliza el siguiente formato de mensaje JSON para el método de serialización Avro:
-
Migración completa
{ "version": 1, "id": 0, "sourceTimestamp": 1702371565, // The timestamp security checkpoint. "sourcePosition": "", // For full migration, the sourcePosition and safeSourcePosition fields are empty. "safeSourcePosition": "", "sourceTxid": "", "source": { "sourceType": "MySQL", // A fixed value, which is MySQL. "version": "OBMySQL" // A fixed value, which is OBMySQL. }, "operation": "INIT", // The operation type for full migration is INIT. "objectName": "test***", "processTimestamps": [ 1702371565238 ], // Only the delivery time is provided. "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" // Only the primary key type is provided. }, "fields": [ { "name": "id", "dataTypeNumber": 246 }, // The data type of each column. { "name": "bid", "dataTypeNumber": 3 }, { "name": "name", "dataTypeNumber": 15 }, { "name": "address", "dataTypeNumber": 254 } ], "beforeImages": null, // The image before full migration is empty. "afterImages": [ // The image after full migration. The precision of a field of the INTEGER or FLOAT type is always 8. The scale is always 64. { "value": "1", "precision": 1, "scale": 0 }, { "precision": 8, "value": "11" }, { "charset": "utf8mb4", "value": { "bytes": "yyy" } }, null ] } -
Sincronización incremental de operaciones DML
-
Ejemplo de inserción de datos (
INSERT){ "version": 1, "id": 170236922143600000, "sourceTimestamp": 1702369092, "sourcePosition": "1702369080", // A checkpoint for the MySQL tenant of OceanBase Database. "safeSourcePosition": "1702369080", // A checkpoint for the MySQL tenant of OceanBase Database. "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, // The image before the INSERT operation is empty. "afterImages": [ {"precision": 8, "value": "2"}, {"precision": 8, "value": "12"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"} } ] } -
Ejemplo de actualización de datos (
UPDATE){ "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": [ // Images before and after the UPDATE operation. {"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"}} ] } -
Ejemplo de eliminación de datos (
DELETE){ "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 // The image after the DELETE operation is empty. }
-
-
Sincronización incremental de operaciones DDL
{ "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, // The values of fields and beforeImages are null for incremental DDL operation synchronization. "beforeImages": null, "afterImages": "alter table multi_db_multi_tbl add column address char(20) default null" // The value of afterImages of the STRING type is a DDL statement. }
Formatos de datos utilizados en la transmisión de datos desde OceanBase Database hacia un sistema de colas de mensajes
Tenga en cuenta los siguientes puntos cuando sincronice datos desde OceanBase Database hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ:
-
Las siguientes tablas describen las asignaciones de tipos de datos para los inquilinos MySQL y Oracle de OceanBase Database cuando el método de serialización es Default, Canal, DataWorks (compatible con DataWorks 2.0), SharePlex o DefaultExtendColumnType.
-
Inquilino MySQL de OceanBase Database
Tipo de dato
Tipo asignado
Descripción
TINYINT
SMALLINT
MEDIUMINT
INT
INTEGER
YEAR
BOOL
BOOLEAN
Long
Enteros con menos de 64 dígitos.
No se utiliza notación científica para valores normales como 1000.
Para valores BOOL o BOOLEAN, true se convierte a 1 y false se convierte a 0.
DECIMAL
NUMERIC
BigDecimal
Numerales decimales exactos y enteros con más de 64 dígitos.
Un entero se muestra sin punto decimal ni cifras decimales.
Un numeral decimal se muestra según los datos pasados a la instancia de base de datos, sin eliminar los ceros al final. Se utiliza notación científica.
FLOAT
DOUBLE
Double
Números de punto flotante.
El número de dígitos significativos se determina según el tipo de dato de origen. Si el tipo de dato de origen es FLOAT, el valor convertido tiene 7 dígitos significativos. Si el tipo de dato de origen es DOUBLE, el valor convertido tiene 16 dígitos significativos.
CHAR
VARCHAR
TINYTEXT
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
SET
String
Tipo cadena.
TINYBLOB
BLOB
MEDIUMBLOB
LONGBLOB
BINARY
VARBINARY
BIT
Bytes
Arreglos de bytes. De forma predeterminada, un arreglo de bytes se muestra como una cadena codificada en Base64. Cuando el método de serialización es Canal, el arreglo de bytes se muestra de forma predeterminada como una cadena codificada en Latin1. En este caso, los datos de tipo BIT se convierten en números.
NotaPara un tipo de dato BIT de longitud fija, la base de datos de destino elimina los ceros en los bits de mayor orden tras recibir un arreglo de bytes durante la sincronización incremental, pero no lo hace durante la sincronización completa. Como resultado, las cadenas codificadas en Base64 mostradas en ambos escenarios pueden ser diferentes. Sin embargo, los resultados reales obtenidos tras la decodificación son consistentes.
DATE
Date
El tipo DATE en formato
YYYY-MM-DD. Si el valor no es válido, se muestra la cadena original.TIME
Time
El tipo TIME en formato
HH:mm:ss[.nnnnnnnnn].La parte de segundos fraccionarios puede contener un máximo de 9 dígitos. Los ceros al final de esta parte no se muestran. Si el valor no es válido, se muestra la cadena original.
DATETIME
DateTime
El tipo DATETIME en formato
YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId], incluida la zona horaria.La parte de segundos fraccionarios puede contener un máximo de 9 dígitos. Los ceros al final de esta parte no se muestran. Si el valor no es válido, se muestra la cadena original.
TIMESTAMP
Timestamp
El tipo TIMESTAMP, en formato
[second-level timestamp][.nnnnnnnnn]. Cuando el método de serialización es Canal, el formato esYYYY-MM-DD HH:mm:ss[.nnnnnnnnn].La parte de segundos fraccionarios puede contener un máximo de 9 dígitos. Los ceros al final de esta parte no se muestran. Si la hora no es válida, se muestra 0000-00-00 00:00:00.
-
Inquilino Oracle de OceanBase Database
Tipo de dato
Tipo asignado
Descripción
INTEGER
Long
Enteros con menos de 64 dígitos.
No se utiliza notación científica para valores normales como 1000.
NUMBER
FLOAT
BigDecimal
Numerales decimales exactos y enteros con más de 64 dígitos.
BINARY_FLOAT BINARY_DOUBLE
Double
Números de punto flotante.
El número de dígitos significativos se determina según el tipo de dato de origen. Si el tipo de dato de origen es FLOAT, el valor convertido tiene 7 dígitos significativos. Si el tipo de dato de origen es DOUBLE, el valor convertido tiene 16 dígitos significativos.
VARCHAR2
NVARCHAR2
INTERVAL YEAR TO MOTH
INTERVAL DAY TO SECOND
CLOB
NCLOB
ROWID
UROWID
String
Tipo cadena.
BLOB
BFILE
RAW
Bytes
Arreglos de bytes.
De forma predeterminada, un arreglo de bytes se muestra como una cadena codificada en Base64.
DATE
TIMESTAMP
TIMESTAMP WITH TIME ZONE
TIMESTAMP WITH LOCAL TIME ZONE
DateTime
El tipo DATETIME en formato
YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId], incluida la zona horaria.La parte de segundos fraccionarios puede contener un máximo de 9 dígitos. Los ceros al final de esta parte no se muestran. Si el valor no es válido, se muestra la cadena original.
-
-
Si el método de serialización es Debezium, las asignaciones de tipos de datos para inquilinos MySQL de OceanBase Database se describen en la siguiente tabla.
ImportanteCuando migra datos desde un inquilino Oracle de OceanBase Database hacia una instancia de Kafka, DataHub (tipo BLOB) o RocketMQ, el método de serialización Debezium no es compatible.
Tipo de dato
Tipo asignado
Descripción
BOOLEAN
BOOL
BOOLEAN
Valores válidos: true y false.
TINYINT
SMALLINT
MEDIUMINT
INT/INTEGER
BIGINT
YEAR
LONG
Enteros en el rango de -263 a 263.
BIGINT
STRING
Cadenas que muestran los datos en su totalidad.
FLOAT
DOUBLE
DOUBLE
Números de punto flotante.
DECIMAL
NUMERIC
STRING
Cadenas que muestran los datos en su totalidad. Un numeral decimal se muestra según los datos pasados a la instancia de base de datos, sin eliminar los ceros al final. Se utiliza notación científica.
BIT
BINARY
VARBINARY
TINYBLOB
BLOB
MEDIUMBLOB
LONGBLOB
BYTES
Arreglos de bytes codificados en Base16.
CHAR
VARCHAR
TINYTEXT
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
SET
STRING
Tipo cadena.
TIMESTAMP
STRING
Cadenas de marca de tiempo en formato YYYY-MM-DDTHH:mm:ss[.nnnnnnnnn]Z. La zona horaria es UTC+0.
DATE
LONG
Número de días transcurridos desde el 1 de enero de 1970.
TIME
LONG
Valor temporal en microsegundos desde 00:00:00, sin incluir la zona horaria.
DATETIME
LONG
Valor temporal en milisegundos desde 1970-01-01 00:00:00, sin incluir la zona horaria.
-
Si el método de serialización es Avro, las asignaciones de tipos de datos para inquilinos MySQL de OceanBase Database se describen en la siguiente tabla.
ImportanteEl método de serialización Avro solo es compatible cuando sincroniza datos desde un inquilino MySQL de OceanBase Database hacia una instancia de Kafka.
Tipo de dato
Tipo asignado
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
NotaPara la migración completa y la sincronización incremental, un valor de tipo TIMESTAMP se convertirá en una marca de tiempo.
-9223372022400Lno es válido.Puede llamar al método
Instant.ofEpochSecond(ts, nanos)de Java para obtener la hora correcta, excepto la hora no válida.DATE
TIME
DATETIME
YEAR
DATETIME
JSON
ENUM
SET
TextObject
GEOMETRY
TextGeometry
NotaActualmente, el servicio de transmisión de datos utiliza el formato EWKT para la transmisión transparente de datos. Por lo tanto, el tipo asignado es TextGeometry.