Todos los productos
Search

Este producto

    DataWorks:Apéndice: formatos de mensaje

    Centro de documentos

    DataWorks:Apéndice: formatos de mensaje

    Última actualización: Oct 10, 2026

    Este tema describe la estructura de los mensajes que se escriben en Kafka y el significado de cada campo.

    Contexto

    Una tarea de sincronización de base de datos completa lee los datos de un origen y los escribe en un topic de Kafka con el formato JSON que se describe a continuación. Cada mensaje contiene la información de las columnas del registro de cambio, así como el estado de los datos antes y después de la modificación. Para que los consumidores de Kafka determinen el avance de la tarea, esta también escribe periódicamente un registro de latido en el topic, donde el campo op se establece en MHEARTBEAT. En las siguientes secciones se describen Overall message format, Synchronization task heartbeat message format y Message format for source data changes. Para conocer los tipos de campo y la descripción de cada parámetro, consulte Field types y Parameter descriptions.

    Formato del mensaje de salida en tiempo real para una sola tabla

    Single-table real-time tasks Al configurar un destino de Kafka, especifique el formato del campo Value en los registros de Kafka. Los formatos compatibles son Canal CDC y JSON. Para más detalles, consulte Appendix: Description of output formats.

    Tipos de campo

    Los datos leídos del origen se asignan a uno de estos seis tipos (BOOLEAN, DOUBLE, DATE, BYTES, LONG y STRING) antes de escribirse en un topic de Kafka con el formato JSON correspondiente.

    Tipo

    Descripción

    BOOLEAN

    Se asigna al tipo booleano de JSON; sus valores son true y false.

    DATE

    Se asigna al tipo numérico de JSON; es una marca de tiempo de 13 dígitos con precisión de milisegundos.

    BYTES

    Se asigna al tipo cadena de JSON. Las matrices de bytes se codifican en Base64 antes de escribirse en Kafka. Los consumidores deben decodificarlas con Base64 (codificación: Base64.getEncoder().encodeToString(text.getBytes("UTF-8")); decodificación: Base64.getDecoder().decode(encodedText)）.

    STRING

    Se asigna al tipo cadena de JSON.

    LONG

    Se asigna al tipo numérico de JSON.

    DOUBLE

    Se asigna al tipo numérico de JSON.

    Descripción de parámetros

    En la siguiente tabla se describen los campos de los mensajes que se escriben en Kafka.

    Elemento de nivel superior

    Elemento de segundo nivel

    Descripción

    schema

    dataColumn

    JSONArray que contiene la información de los tipos de columna de datos.dataColumnRegistra todas las columnas y sus tipos en los registros de cambio del origen. Estos cambios incluyen inserciones, eliminaciones o actualizaciones de datos, así como modificaciones en el esquema de las tablas de la base de datos.

    • name: nombre de la columna.

    • type: tipo de la columna.

    primaryKey

    Lista que contiene la información de la clave primaria.

    pk: nombre de la clave primaria.

    source

    Objeto que contiene la información de la base de datos o tabla de origen.

    • dbType: String, tipo de base de datos.

    • dbVersion: String, versión de la base de datos.

    • dbName: String, nombre de la base de datos.

    • schemaName: String, nombre del esquema (para Postgres, SQL Server y bases de datos similares).

    • tableName: String, nombre de la tabla.

    payload

    before

    JSONObject que contiene los datos previos a una actualización. Por ejemplo, si un origen MySQL actualiza un registro, before almacena los datos de ese registro antes de la modificación.

    • Este campo se llena en los mensajes de actualización y eliminación leídos del origen.

    • dataColumn: JSONObject que contiene los datos con el formato nombre de columna: valor de columna. Los nombres de columna son cadenas de texto; los valores pueden ser BOOLEAN, DOUBLE, DATE, BYTES, LONG o STRING.

    after

    Datos posteriores a la actualización, con el mismo formato que before.

    sequenceId

    Cadena generada por Streamx para ordenar los datos incrementales y completos durante una fusión. Cada registro de Streamx tiene un sequenceId único.

    Nota

    Un mensaje de actualización leído del origen genera dos registros de salida (update before y update after) con el mismo sequenceId.

    scn

    Válido cuando el origen es Oracle; contiene el SCN de Oracle.

    op

    El tipo de mensaje leído del origen. Valores válidos:

    • INSERT: datos insertados.

    • UPDATE_BEFOR: estado previo a la actualización de datos.

    • UPDATE_AFTER: estado posterior a la actualización de datos.

    • DELETE: datos eliminados.

    • TRANSACTION_BEGIN: inicio de una transacción de base de datos.

    • TRANSACTION_END: fin de una transacción de base de datos.

    • CREATE: tabla de base de datos creada.

    • ALTER: tabla de base de datos modificada.

    • QUERY: instrucción SQL original de un cambio en la base de datos.

    • TRUNCATE: tabla de base de datos vaciada.

    • RENAME: tabla de base de datos renombrada.

    • CINDEX: índice creado.

    • DINDEX: índice eliminado.

    • MHEARTBEAT: latido que indica que la sincronización avanza con normalidad cuando no hay datos nuevos disponibles en el origen.

    timestamp

    JSONObject que contiene las marcas de tiempo relacionadas con este registro.

    • eventTime: Long, momento en que cambió la base de datos de origen, expresado como una marca de tiempo de 13 dígitos en milisegundos.

    • systemTime: Long, momento en que la tarea procesó este mensaje de cambio, expresado como una marca de tiempo de 13 dígitos en milisegundos.

    • checkpointTime: Long, hora establecida al restablecer el punto de control de sincronización, expresada como una marca de tiempo de 13 dígitos en milisegundos. Suele coincidir con eventTime.

    ddl

    Este campo solo se llena cuando cambia el esquema de una tabla de la base de datos. Para los cambios de datos (inserciones, eliminaciones y actualizaciones), el campo ddl se establece en null.

    • text: String, la instrucción DDL de la base de datos.

    • ddlMeta: String, una cadena codificada en Base64 que se obtiene al serializar un objeto Java donde se registra el tipo de cambio DDL de la base de datos.

    version

    Ninguna

    Versión del formato de mensaje.

    Formato general del mensaje

    El formato general de un mensaje escrito en Kafka es el siguiente:

    { "schema": { //Metadatos del cambio; solo contiene nombres y tipos de columna "dataColumn": [//Información de las columnas de datos modificadas para actualizar el registro de la tabla de destino { "name": "id", "type": "LONG" }, { "name": "name", "type": "STRING" }, { "name": "binData", "type": "BYTES" }, { "name": "ts", "type": "DATE" }, { "name":"rowid",// Si el origen es Oracle, rowid se incluye entre las columnas de datos "type":"STRING" } ], "primaryKey": [ "pkName1", "pkName2" ], "source": { "dbType": "mysql", "dbVersion": "1.0.0", "dbName": "myDatabase", "schemaName": "mySchema", "tableName": "tableName" } }, "payload": { "before": { "dataColumn":{ "id": 111, "name":"scooter", "binData": "[base64 string]", "ts": 1590315269000, "rowid": "AAIUMPAAFAACxExAAE"//Cadena que contiene el rowid de Oracle } }, "after": { "dataColumn":{ "id": 222, "name":"donald", "binData": "[base64 string]", "ts": 1590315269000, "rowid": "AAIUMPAAFAACxExAAE"//Cadena que contiene el rowid de Oracle } }, "sequenceId":"XXX",//Cadena utilizada para ordenar los datos al fusionar información incremental y completa, "scn":"xxxx",//Cadena que contiene el SCN de Oracle "op": "INSERT/UPDATE_BEFOR/UPDATE_AFTER/DELETE/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/MHEARTBEAT...",//Distingue entre mayúsculas y minúsculas, "timestamp": { "eventTime": 1,//Obligatorio: momento en que cambió la base de datos de origen, como marca de tiempo de 13 dígitos en milisegundos "systemTime": 2,//Opcional: momento en que la tarea procesó el cambio, como marca de tiempo de 13 dígitos en milisegundos "checkpointTime": 3//Opcional: hora establecida al restablecer el punto de control de sincronización, como marca de tiempo de 13 dígitos en milisegundos; suele coincidir con eventTime }, "ddl": { "text": "ADD COLUMN ...", "ddlMeta": "[SQLStatement serialized binary, expressed in base64 string]" } }, "version":"1.0.0" }

    Nota

    Para conocer los tipos de campo y la descripción de cada parámetro, consulte Field types y Parameter descriptions.

    Formato del mensaje de latido de la tarea de sincronización

    {
    "schema": {
        "dataColumn": null,
        "primaryKey": null,
        "source": null
    },
    "payload": {
        "before": null,
        "after": null,
        "sequenceId": null,
        "timestamp": {
            "eventTime": 1620457659000,
            "checkpointTime": 1620457659000
        },
        "op": "MHEARTBEAT",
        "ddl": null
    },
    "version": "0.0.1"
}
    Nota

    Para conocer los tipos de campo y la descripción de cada parámetro, consulte Field types y Parameter descriptions.

    Formato de mensaje para cambios en los datos de origen

    • Formato del mensaje de Kafka para una inserción en el origen:

      {
    "schema": {
        "dataColumn": [
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "job",
                "type": "STRING"
            },
            {
                "name": "sex",
                "type": "STRING"
            },
            {
                "name": "#alibaba_rds_row_id#",
                "type": "LONG"
            }
        ],
        "primaryKey": null,
        "source": {
            "dbType": "MySQL",
            "dbName": "pkset_test",
            "tableName": "pkset_test_no_pk"
        }
    },
    "payload": {
        "before": null,
        "after": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "man",
                "#alibaba_rds_row_id#": 15
            }
        },
        "sequenceId": "1620457642589000000",
        "timestamp": {
            "eventTime": 1620457896000,
            "systemTime": 1620457896977,
            "checkpointTime": 1620457896000
        },
        "op": "INSERT",
        "ddl": null
    },
    "version": "0.0.1"
}

    • Formato del mensaje de Kafka para una actualización en el origen:

      • Si no se selecciona When one record in the source is updated, one Kafka record is generated., una actualización en el origen genera dos mensajes de Kafka: uno para el estado previo y otro para el estado posterior. Sus formatos son los siguientes:

        Formato del mensaje para el estado previo a la actualización:

        {
    "schema": {
        "dataColumn": [
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "job",
                "type": "STRING"
            },
            {
                "name": "sex",
                "type": "STRING"
            },
            {
                "name": "#alibaba_rds_row_id#",
                "type": "LONG"
            }
        ],
        "primaryKey": null,
        "source": {
            "dbType": "MySQL",
            "dbName": "pkset_test",
            "tableName": "pkset_test_no_pk"
        }
    },
    "payload": {
        "before": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "man",
                "#alibaba_rds_row_id#": 15
            }
        },
        "after": null,
        "sequenceId": "1620457642589000001",
        "timestamp": {
            "eventTime": 1620458077000,
            "systemTime": 1620458077779,
            "checkpointTime": 1620458077000
        },
        "op": "UPDATE_BEFOR",
        "ddl": null
    },
    "version": "0.0.1"
}

        Formato del mensaje para el estado posterior a la actualización:

        {
    "schema": {
        "dataColumn": [
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "job",
                "type": "STRING"
            },
            {
                "name": "sex",
                "type": "STRING"
            },
            {
                "name": "#alibaba_rds_row_id#",
                "type": "LONG"
            }
        ],
        "primaryKey": null,
        "source": {
            "dbType": "MySQL",
            "dbName": "pkset_test",
            "tableName": "pkset_test_no_pk"
        }
    },
    "payload": {
        "before": null,
        "after": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "woman",
                "#alibaba_rds_row_id#": 15
            }
        },
        "sequenceId": "1620457642589000001",
        "timestamp": {
            "eventTime": 1620458077000,
            "systemTime": 1620458077779,
            "checkpointTime": 1620458077000
        },
        "op": "UPDATE_AFTER",
        "ddl": null
    },
    "version": "0.0.1"
}

      • Si se selecciona When one record in the source is updated, one Kafka record is generated., una actualización en el origen genera un solo mensaje de Kafka que contiene tanto el estado previo como el posterior. Su formato es el siguiente:

        {
    "schema": {
        "dataColumn": [
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "job",
                "type": "STRING"
            },
            {
                "name": "sex",
                "type": "STRING"
            },
            {
                "name": "#alibaba_rds_row_id#",
                "type": "LONG"
            }
        ],
        "primaryKey": null,
        "source": {
            "dbType": "MySQL",
            "dbName": "pkset_test",
            "tableName": "pkset_test_no_pk"
        }
    },
    "payload": {
        "before": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "man",
                "#alibaba_rds_row_id#": 15
            }
        },
        "after": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "woman",
                "#alibaba_rds_row_id#": 15
            }
        },
        "sequenceId": "1620457642589000001",
        "timestamp": {
            "eventTime": 1620458077000,
            "systemTime": 1620458077779,
            "checkpointTime": 1620458077000
        },
        "op": "UPDATE_AFTER",
        "ddl": null
    },
    "version": "0.0.1"
}

    • Formato del mensaje de Kafka para una eliminación en el origen:

      {
    "schema": {
        "dataColumn": [
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "job",
                "type": "STRING"
            },
            {
                "name": "sex",
                "type": "STRING"
            },
            {
                "name": "#alibaba_rds_row_id#",
                "type": "LONG"
            }
        ],
        "primaryKey": null,
        "source": {
            "dbType": "MySQL",
            "dbName": "pkset_test",
            "tableName": "pkset_test_no_pk"
        }
    },
    "payload": {
        "before": {
            "dataColumn": {
                "name": "name11",
                "job": "job11",
                "sex": "woman",
                "#alibaba_rds_row_id#": 15
            }
        },
        "after": null,
        "sequenceId": "1620457642589000002",
        "timestamp": {
            "eventTime": 1620458266000,
            "systemTime": 1620458266101,
            "checkpointTime": 1620458266000
        },
        "op": "DELETE",
        "ddl": null
    },
    "version": "0.0.1"
}
    Nota

    Para conocer los tipos de campo y la descripción de cada parámetro, consulte Field types y Parameter descriptions.