Data storage formats in message queues - Data Transmission Service

Data Transmission Service (DTS) le permite seleccionar un formato de almacenamiento cuando sincroniza o migra datos a una cola de mensajes, como Kafka o RocketMQ. Este tema describe los formatos de datos para ayudarle a analizar sintácticamente la información.

Formatos de almacenamiento de datos

DTS admite los siguientes tres formatos de almacenamiento para los datos que se escriben en una cola de mensajes:

DTS Avro: formato de serialización de datos que convierte las estructuras u objetos de datos en un formato fácil de almacenar o transmitir.
Shareplex Json: cuando el software de replicación de datos SharePlex lee datos de una base de datos de origen y los escribe en una cola de mensajes, estos se almacenan en formato Shareplex Json.
Canal Json: Canal analiza sintácticamente los registros incrementales de una base de datos y transmite los datos incrementales a una cola de mensajes. Los datos se almacenan en formato Canal Json.

DTS Avro

Debe analizar sintácticamente los datos con base en la definición del esquema DTS Avro. Para obtener más información, consulte Definición del esquema DTS Avro y Ejemplo de deserialización DTS Avro.

Nota

En el formato DTS Avro, las instrucciones DDL son de tipo String.

Shareplex Json

Descripción de parámetros

Parámetro	Descripción
`time`	Hora en la que se confirmó la transacción en la base de datos. El formato es yyyy-MM-ddTHH:mm:ssZ (UTC).
`userid`	Identificador del usuario que confirmó la transacción.
`op`	Tipo de operación de datos. Los valores válidos incluyen INSERT, UPDATE, DELETE, TRUNCATE, DROP COLUMN, UPDATE BEFORE y UPDATE AFTER.
`scn`	Número de cambio del sistema (SCN). Identifica la versión de una transacción confirmada en un momento específico dentro de la base de datos. A cada transacción confirmada se le asigna un SCN único.
`rowid`	Valor de dirección relativamente único que se utiliza para localizar un registro en la base de datos.
`trans`	Identificador de la transacción.
`seq`	Número ordinal de la operación dentro de la transacción. El valor comienza en 1.
`size`	Número total de operaciones en la transacción.
`table`	Nombre de la tabla.
`idx`	Índice de la operación dentro de la transacción. El formato es `seq/size`. Por ejemplo, `1/11` indica que esta es la primera operación en una transacción que contiene 11 operaciones.
`posttime`	Hora en la que se confirmó la transacción en la base de datos de destino.

Ejemplos

Insertar datos

{
    "meta": {
        "time": "2017-06-16T14:24:34", 
        "userid": 84,                                    
        "op": "ins",                                   
          "scn": "14589063118712",                  
          "rowid": "AAATGpAAIAAItcIAAA",      
        "trans": "7.0.411499",                 
        "seq": 1,                                          
        "size": 11,                                         
        "table": "CL_BIZ1.MIO_LOG",       
          "idx": "1/11",                                       
        "posttime": "2017-06-16T14:33:52"
    },
    "data": {
        "MIO_LOG_ID": "32539737"
     }
}

Actualizar datos

{
    "meta": {
        "time": "2017-06-16T15:38:13",
        "userid": 84,
        "op": "upd",                             
        "table": "CL_BIZ1.MIO_LOG"
        ….
    },
    "data": {                                          
        "CNTR_NO": "1171201606"
    },
    "key": {                                            
        "MIO_LOG_ID": "32537893",
        "PLNMIO_REC_ID": "31557806",
        "POL_CODE": null,
        "CNTR_TYPE": null,
        "CNTR_NO": "1171201606syui26"
    }
}

Eliminar datos

{
    "meta": {
        "time": "2017-06-16T15:51:35",
        "userid": 84,
        "op": "del",                      
     },
    "data": {                                    
        "MIO_LOG_ID": "32539739",
        "PLNMIO_REC_ID": "31557806",
        "POL_CODE": null,
        "CNTR_TYPE": null,
        "CG_NO": null
     }
}

Canal Json

Nota

Si habilita Split message delivery after partition key update, Kafka entrega un mensaje DELETE y un mensaje INSERT cuando se modifica una clave de partición. Kafka selecciona la partición de cada mensaje según el valor de su respectiva clave de partición.

Ejemplo: la clave de partición id tiene un valor de 1. El mensaje se entrega a partition-1. A continuación se describen las diferencias antes y después de habilitar Split message delivery after partition key update.

Habilitado: cuando ejecuta el comando UPDATE SET id = 2 WHERE id = 1 en la base de datos de origen, se entrega un mensaje DELETE con id=1 a partition-1, y un mensaje INSERT con id=2 a partition-2.
Deshabilitado: solo se entrega un mensaje UPDATE a partition-1. La operación UPDATE selecciona la partición de entrega con base en el valor anterior al cambio.

Descripción de métricas

Parámetro	Descripción
`database`	Nombre de la base de datos.
`es`	Hora en la que se ejecutó la operación en la base de datos de origen. Es una marca de tiempo UNIX de 13 dígitos en milisegundos. Nota El tiempo real de ejecución está en segundos. Se agregan tres ceros para convertir la unidad a milisegundos. Puede usar un motor de búsqueda para encontrar una herramienta de conversión de marcas de tiempo UNIX.
`id`	Número de serie de la operación. Nota Se genera a partir de la marca de tiempo y un desplazamiento interno de DTS. Le ayuda a determinar el orden de los registros.
`isDdl`	Indica si la operación es de tipo DDL. true: sí. El valor false significa que no.
`mysqlType`	Tipo de datos del campo. Nota No se admiten los parámetros de tipos de datos, como la precisión.
`old` y `data`	Datos anteriores o posteriores al cambio. Nota Para las instancias de sincronización o migración creadas antes del 20 de marzo de 2022, el valor de old corresponde a los datos posteriores al cambio, mientras que el valor de `data` corresponde a los datos anteriores al cambio. De forma predeterminada, el valor de `old` incluye todas las columnas, no solo las modificadas. Para alinearse con la comunidad de código abierto, en las instancias creadas o reiniciadas a partir del 20 de marzo de 2022, el valor de `data` corresponde a los datos posteriores al cambio, y el valor de `old` corresponde a los datos anteriores al cambio.
`pkNames`	Nombre de la clave principal.
`sql`	Instrucción SQL.
`sqlType`	Tipo de campo transformado. El valor es igual al de dataTypeNumber. Para obtener más información, consulte Mapping between field types and dataTypeNumber values.
`table`	Nombre de la tabla.
`ts`	Hora en la que la operación comenzó a escribir datos en la base de datos de destino. Es una marca de tiempo UNIX de 13 dígitos en milisegundos. Nota Puede usar un motor de búsqueda para encontrar una herramienta de conversión de marcas de tiempo UNIX.
`type`	Tipo de operación, como DELETE, UPDATE o INSERT. Nota Para las tareas de sincronización o migración de datos completos, el valor se fija en INIT.
`gtid`	Identificador global de transacciones (GTID). Un GTID es único a nivel global. Cada transacción corresponde a un GTID. Nota DTS no admite la sincronización del campo GTID. El valor de este campo es NULL de forma predeterminada.

Ejemplos

Actualizar datos

Nota

Para las instancias de sincronización o migración creadas antes del 20 de marzo de 2022, cuando una instrucción DELETE de una tabla de origen se sincroniza o migra a Kafka, el campo old contiene los datos y el campo data es nulo. Para alinearse con la comunidad de código abierto, en las instancias creadas o reiniciadas a partir del 20 de marzo de 2022, el campo data contiene los datos y el campo old es nulo.

Instancias de sincronización o migración creadas antes del 20 de marzo de 2022

{
    "old": [
        {
            "shipping_type": "aaa"
        }
    ], 
    "database": "dbname", 
    "es": 1600161894000, 
    "id": 58, 
    "isDdl": false, 
    "mysqlType": {
        "id": "bigint", 
        "shipping_type": "varchar"
    }, 
    "pkNames": [
        "id"
    ], 
    "sql": "", 
    "sqlType": {
        "id": -5, 
        "shipping_type": 12
    }, 
    "table": "tablename", 
    "ts": 1600161894771, 
    "type": "DELETE"
}

Instancias de sincronización o migración creadas o reiniciadas a partir del 20 de marzo de 2022

{
    "data": [
        {
            "id": "500000287", 
            "shipping_type": null
        }
    ], 
    "database": "dbname", 
    "es": 1600161894000, 
    "id": 58, 
    "isDdl": false, 
    "mysqlType": {
        "id": "bigint", 
        "shipping_type": "varchar"
    }, 
    "pkNames": [
        "id"
    ], 
    "sql": "", 
    "sqlType": {
        "id": -5, 
        "shipping_type": 12
    }, 
    "table": "tablename", 
    "ts": 1600161894771, 
    "type": "DELETE"
}

Operación DDL

{
    "database":"dbname", // The name of the database for synchronization or migration.
    "es":1600161894000, // The time when the source data was written to the binary log.
    "id":58, // The offset in the DTS cache.
    "isDdl":true, // Specifies whether to synchronize or migrate DDL statements.
    "sql":"eg:createxxx", // The DDL statement from the binary log.
    "table":"tablename", // The name of the table for synchronization or migration.
    "ts":1600161894771, // The time when DTS wrote the data to the destination.
    "type":"DDL"
}