Format of messages written to Kafka - DataWorks: Central de Documentação do Alibaba Cloud

Contexto

Uma tarefa de sincronização completa de banco de dados lê dados de uma source upstream e os grava em um tópico do Kafka no formato JSON descrito abaixo. Uma mensagem contém informações de coluna para o registro de alteração e os estados dos dados antes e depois da mudança. Para que os consumidores do Kafka determinem o progresso da tarefa, ela também grava periodicamente um registro de heartbeat no tópico com o campo op defina como MHEARTBEAT. As seções a seguir descrevem o Formato geral da mensagem, o Formato de mensagem de heartbeat da tarefa de sincronização e o Formato de mensagem para alterações de dados da source. Para tipos de campo e descrições de parâmetros, consulte Tipos de campo e Descrições de parâmetros.

Formato de mensagem de saída em tempo real de tabela única

Single-table real-time tasks Ao configurar um destino Kafka, especifique o formato do campo Value nos registros do Kafka. Há suporte para Canal CDC e JSON. Para mais detalhes, consulte Appendix: Description of output formats.

Tipos de campo

Os dados lidos da source são mapeados para um dos seis tipos (BOOLEAN, DOUBLE, DATE, BYTES, LONG e STRING) antes da gravação em um tópico do Kafka no formato JSON aplicável.

Tipo	Descrição
BOOLEAN	Mapeia para o tipo Boolean do JSON; os valores são true e false.
DATE	Mapeia para o tipo number do JSON; um timestamp de 13 dígitos com precisão de milissegundos.
BYTES	Mapeia para o tipo string do JSON. Arrays de bytes são codificados em Base64 antes da gravação no Kafka, e os consumidores devem decodificá-los em Base64 (codificação: Base64.getEncoder().encodeToString(text.getBytes("UTF-8")); decodificação: Base64.getDecoder().decode(encodedText)）.
STRING	Mapeia para o tipo string do JSON.
LONG	Mapeia para o tipo number do JSON.
DOUBLE	Mapeia para o tipo number do JSON.

Descrições de parâmetros

A tabela a seguir descreve os campos das mensagens gravadas no Kafka.

Elemento de nível superior	Elemento de segundo nível	Descrição
schema	dataColumn	JSONArray que contém informações sobre o tipo das colunas de dados. O dataColumn registra todas as colunas e seus tipos nos registros de alteração da source. As alterações incluem inserção, exclua ou atualize de dados, além de mudanças nos schemas das tabelas do banco de dados. name: nome da coluna type: tipo da coluna
	primaryKey	Lista que contém informações da chave primária. pk: nome da chave primária.
	source	Objeto que contém informações do banco de dados ou tabela de source. dbType: String, tipo do banco de dados. dbVersion: String, versão do banco de dados. dbName: String, nome do banco de dados. schemaName: String, nome do schema (para Postgres, SQL Server e bancos de dados semelhantes). tableName: String, nome da tabela.
payload	before	JSONObject que contém os dados antes de uma atualize. Por exemplo, se uma source MySQL atualize um registro, before contém os dados do registro antes da atualize. Este campo é preenchido para mensagens de atualize e exclua lidas da source. dataColumn: JSONObject que contém os dados no formato nome da coluna: valor da coluna. Os nomes das colunas são strings; os valores podem ser BOOLEAN, DOUBLE, DATE, BYTES, LONG ou STRING.
	after	Dados após a atualize, no mesmo formato de before.
	sequenceId	String gerada pelo Streamx para ordenar dados incrementais e completos durante uma mesclagem. Cada registro do Streamx possui um sequenceId exclusivo. Nota Uma mensagem de atualize lida da source produz dois registros de saída, update before e update after, com o mesmo sequenceId.
	scn	Válido quando a source é Oracle; contém o SCN do Oracle.
	op	O tipo de mensagem lido da source. Valores válidos: INSERT: dados inseridos. UPDATE_BEFOR: antes do atualize de dados. UPDATE_AFTER: após o atualize de dados. DELETE: dados exclua. TRANSACTION_BEGIN: início da transação do banco de dados. TRANSACTION_END: fim da transação do banco de dados. CREATE: tabela do banco de dados crie. ALTER: tabela do banco de dados alterada. QUERY: SQL original de uma alteração no banco de dados. TRUNCATE: tabela do banco de dados limpa. RENAME: tabela do banco de dados renomeada. CINDEX: índice criado. DINDEX: índice exclua. MHEARTBEAT: heartbeat indicando que a sincronização ocorre normalmente quando não há novos dados da source disponíveis.
	timestamp	JSONObject que contém timestamps relacionados a este registro. eventTime: Long, momento em que o banco de dados de source foi alterado, em um timestamp de 13 dígitos em milissegundos. systemTime: Long, momento em que a tarefa processou esta mensagem de alteração, em um timestamp de 13 dígitos em milissegundos. checkpointTime: Long, horário defina ao redefinir o checkpoint de sincronização, em um timestamp de 13 dígitos em milissegundos. Geralmente é igual a eventTime.
	ddl	Este campo é preenchido apenas para alterações de schema de tabela do banco de dados. Para alterações de dados, incluindo inserções, exclua e atualize, o campo ddl é defina como null. text: String, a instrução DDL do banco de dados. ddlMeta: String, uma string codificada em Base64 produzida pela serialização de um objeto Java que registra o tipo de alteração DDL do banco de dados.
version	Nenhum	Versão do formato de mensagem.

Formato geral da mensagem

O formato geral de uma mensagem gravada no Kafka é:

{ "schema": { //Metadados de alteração, contendo apenas nomes e tipos de colunas "dataColumn": [//Informações da coluna de dados alterada para atualize o registro da tabela de destino { "name": "id", "type": "LONG" }, { "name": "name", "type": "STRING" }, { "name": "binData", "type": "BYTES" }, { "name": "ts", "type": "DATE" }, { "name":"rowid",// Para uma source Oracle, rowid é incluído entre as colunas de dados "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 que contém o rowid do Oracle } }, "after": { "dataColumn":{ "id": 222, "name":"donald", "binData": "[base64 string]", "ts": 1590315269000, "rowid": "AAIUMPAAFAACxExAAE"//String que contém o rowid do Oracle } }, "sequenceId":"XXX",//String usada para ordenar dados ao mesclar dados incrementais e completos, "scn":"xxxx",//String que contém o SCN do Oracle "op": "INSERT/UPDATE_BEFOR/UPDATE_AFTER/DELETE/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/MHEARTBEAT...",//Sensível a maiúsculas e minúsculas "timestamp": { "eventTime": 1,//Obrigatório: momento em que o banco de dados de source foi alterado, em um timestamp de 13 dígitos em milissegundos "systemTime": 2,//Opcional: momento em que a tarefa processou a alteração, em um timestamp de 13 dígitos em milissegundos "checkpointTime": 3//Opcional: horário defina ao redefinir o checkpoint de sincronização, em um timestamp de 13 dígitos em milissegundos, geralmente igual a eventTime }, "ddl": { "text": "ADD COLUMN ...", "ddlMeta": "[Binário serializado de SQLStatement, expresso em string base64]" } }, "version":"1.0.0" }

Nota

Para tipos de campo e descrições de parâmetros, consulte Tipos de campo e Descrições de parâmetros.

Formato de mensagem de heartbeat da tarefa de sincronização

{
    "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 tipos de campo e descrições de parâmetros, consulte Tipos de campo e Descrições de parâmetros.

Formato de mensagem para alterações de dados da source

Formato de mensagem do Kafka para uma inserção na 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"
}

Formato de mensagem do Kafka para uma atualize na source:

Quando When one record in the source is updated, one Kafka record is generated. não estiver selecionada, uma atualize na source produz duas mensagens do Kafka, uma para o estado antes da atualize e outra para o estado após a atualize. Os formatos são:

Formato de mensagem para o estado antes da atualize:

{
    "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 de mensagem para o estado após a atualize:

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

Quando When one record in the source is updated, one Kafka record is generated. estiver selecionada, uma atualize na source produz uma mensagem do Kafka contendo os estados anterior e posterior. O formato é:

{
    "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 de mensagem do Kafka para uma exclua na 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"
}

Nota

Para tipos de campo e descrições de parâmetros, consulte Tipos de campo e Descrições de parâmetros.