Format des messages écrits dans Kafka - DataWorks - Centre de documentation Alibaba Cloud

Contexte

Une tâche de synchronisation complète de base de données lit les données depuis une source en amont et les écrit dans un topic Kafka au format JSON décrit ci-dessous. Un message contient les informations de colonne pour l'enregistrement de modification ainsi que les états des données avant et après la modification. Pour permettre aux consommateurs Kafka de déterminer la progression de la tâche, celle-ci écrit également périodiquement un enregistrement de heartbeat dans le topic, avec le champ op défini sur MHEARTBEAT. Les sections suivantes décrivent le Format global des messages, le Format des messages de heartbeat des tâches de synchronisation et le Format des messages pour les modifications de données source. Pour les types de champs et les descriptions des paramètres, consultez les sections Types de champs et Descriptions des paramètres.

Format des messages de sortie en temps réel pour une table unique

Lors de la configuration d'une destination Kafka pour les tâches en temps réel pour une table unique, spécifiez le format du champ Value dans les enregistrements Kafka. Les formats Canal CDC et JSON sont pris en charge. Pour plus de détails, consultez l'Annexe : Description des formats de sortie.

Types de champs

Les données lues depuis la source sont mappées vers l'un des six types suivants — BOOLEAN, DOUBLE, DATE, BYTES, LONG et STRING — avant d'être écrites dans un topic Kafka au format JSON applicable.

Type

Description

BOOLEAN

Correspond au type Boolean JSON ; les valeurs sont true et false.

DATE

Correspond au type number JSON ; un horodatage de 13 chiffres précis à la milliseconde.

BYTES

Correspond au type string JSON. Les tableaux d'octets sont encodés en Base64 avant d'être écrits dans Kafka et doivent être décodés en Base64 par les consommateurs (encodage : Base64.getEncoder().encodeToString(text.getBytes("UTF-8")) ; décodage : Base64.getDecoder().decode(encodedText)）.

STRING

Correspond au type string JSON.

LONG

Correspond au type number JSON.

DOUBLE

Correspond au type number JSON.

Descriptions des paramètres

Le tableau suivant décrit les champs des messages écrits dans Kafka.

Élément de premier niveau	Élément de deuxième niveau	Description
schema	dataColumn	JSONArray contenant les informations de type des colonnes de données. dataColumn enregistre toutes les colonnes et leurs types dans les enregistrements de modification en amont. Les modifications incluent l'insertion, la suppression ou la mise à jour des données, ainsi que les modifications des schémas de table de base de données. name : nom de la colonne type : type de la colonne
	primaryKey	Liste contenant les informations de clé primaire. pk : nom de la clé primaire.
	source	Objet contenant les informations de la base de données ou de la table source. dbType : String, type de base de données. dbVersion : String, version de la base de données. dbName : String, nom de la base de données. schemaName : String, nom du schéma (pour les bases de données Postgres, SQL Server et similaires). tableName : String, nom de la table.
payload	before	JSONObject contenant les données avant une mise à jour. Par exemple, si une source MySQL met à jour un enregistrement, before contient les données de l'enregistrement avant la mise à jour. Ce champ est renseigné pour les messages de mise à jour et de suppression lus depuis la source. dataColumn : JSONObject contenant les données au format nom de la colonne : valeur de la colonne. Les noms de colonne sont des chaînes ; les valeurs peuvent être BOOLEAN, DOUBLE, DATE, BYTES, LONG ou STRING.
	after	Données après la mise à jour, dans le même format que before.
	sequenceId	String générée par Streamx pour ordonner les données incrémentielles et complètes lors d'une fusion. Chaque enregistrement Streamx possède un sequenceId unique. Remarque Un message de mise à jour lu depuis la source produit deux enregistrements de sortie, update before et update after, avec le même sequenceId.
	scn	Valide lorsque la source est Oracle ; contient le SCN Oracle.
	op	Le type de message lu depuis la source. Valeurs valides : INSERT : données insérées. UPDATE_BEFOR : avant une mise à jour des données. UPDATE_AFTER : après une mise à jour des données. DELETE : données supprimées. TRANSACTION_BEGIN : début d'une transaction de base de données. TRANSACTION_END : fin d'une transaction de base de données. CREATE : création d'une table de base de données. ALTER : modification d'une table de base de données. QUERY : SQL original pour une modification de base de données. TRUNCATE : vidage d'une table de base de données. RENAME : renommage d'une table de base de données. CINDEX : création d'un index. DINDEX : suppression d'un index. MHEARTBEAT : heartbeat indiquant que la synchronisation se déroule normalement lorsqu'aucune nouvelle donnée source n'est disponible.
	timestamp	JSONObject contenant les horodatages liés à cet enregistrement. eventTime : Long, moment où la base de données source a été modifiée, sous forme d'horodatage en millisecondes de 13 chiffres. systemTime : Long, moment où la tâche a traité ce message de modification, sous forme d'horodatage en millisecondes de 13 chiffres. checkpointTime : Long, heure définie lors de la réinitialisation du point de contrôle de synchronisation, sous forme d'horodatage en millisecondes de 13 chiffres. Généralement égal à eventTime.
	ddl	Ce champ est renseigné uniquement pour les modifications de schéma de table de base de données. Pour les modifications de données, y compris les insertions, suppressions et mises à jour, le champ ddl est défini sur null. text : String, l'instruction DDL de la base de données. ddlMeta : String, une chaîne encodée en Base64 produite par la sérialisation d'un objet Java qui enregistre le type de modification DDL de la base de données.
version	Aucun	Version du format de message.

Format global des messages

Le format global d'un message écrit dans Kafka est le suivant :

{ "schema": { //Métadonnées de modification, contenant uniquement les noms et types de colonnes "dataColumn": [//Informations sur les colonnes de données modifiées pour la mise à jour de l'enregistrement de la table de destination { "name": "id", "type": "LONG" }, { "name": "name", "type": "STRING" }, { "name": "binData", "type": "BYTES" }, { "name": "ts", "type": "DATE" }, { "name":"rowid",// Pour une source Oracle, rowid est inclus parmi les colonnes de données "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"//String contenant le rowid Oracle } }, "after": { "dataColumn":{ "id": 222, "name":"donald", "binData": "[base64 string]", "ts": 1590315269000, "rowid": "AAIUMPAAFAACxExAAE"//String contenant le rowid Oracle } }, "sequenceId":"XXX",//String utilisée pour ordonner les données lors de la fusion des données incrémentielles et complètes, "scn":"xxxx",//String contenant le SCN Oracle "op": "INSERT/UPDATE_BEFOR/UPDATE_AFTER/DELETE/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/MHEARTBEAT...",//Sensible à la casse, "timestamp": { "eventTime": 1,//Requis : moment où la base de données source a été modifiée, sous forme d'horodatage en millisecondes de 13 chiffres "systemTime": 2,//Facultatif : moment où la tâche a traité la modification, sous forme d'horodatage en millisecondes de 13 chiffres "checkpointTime": 3//Facultatif : heure définie lors de la réinitialisation du point de contrôle de synchronisation, sous forme d'horodatage en millisecondes de 13 chiffres, généralement égal à eventTime }, "ddl": { "text": "ADD COLUMN ...", "ddlMeta": "[SQLStatement binaire sérialisé, exprimé en chaîne base64]" } }, "version":"1.0.0" }

Remarque

Pour les types de champs et les descriptions des paramètres, consultez les sections Types de champs et Descriptions des paramètres.

Format des messages de heartbeat des tâches de synchronisation

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

Remarque

Pour les types de champs et les descriptions des paramètres, consultez les sections Types de champs et Descriptions des paramètres.

Format des messages pour les modifications de données source

Format des messages Kafka pour une insertion source :

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

Format des messages Kafka pour une mise à jour source :

Si l'option When one record in the source is updated, one Kafka record is generated. n'est pas sélectionnée, une mise à jour source produit deux messages Kafka, un pour l'état avant la mise à jour et un pour l'état après. Les formats sont les suivants :

Format du message pour l'état avant la mise à jour :

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

Format du message pour l'état après la mise à jour :

{
    "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 l'option When one record in the source is updated, one Kafka record is generated. est sélectionnée, une mise à jour source produit un seul message Kafka contenant à la fois les états avant et après. Le format est le suivant :

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

Format des messages Kafka pour une suppression source :

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

Remarque

Pour les types de champs et les descriptions des paramètres, consultez les sections Types de champs et Descriptions des paramètres.