All Products
Search

This Product

    DataWorks:Lampiran: Format pesan

    Document Center

    DataWorks:Lampiran: Format pesan

    Last Updated:May 05, 2026

    Topik ini menjelaskan struktur dan bidang pesan yang ditulis ke Kafka.

    Informasi latar belakang

    Saat Anda menyinkronkan seluruh database ke Kafka, tugas sinkronisasi membaca data dari sumber data hulu dan menuliskannya ke topik Kafka dalam format JSON yang dijelaskan dalam topik ini. Format pesan secara keseluruhan mencakup informasi kolom untuk catatan perubahan serta status data sebelum dan sesudah perubahan. Untuk memungkinkan konsumen melacak progres tugas, tugas sinkronisasi juga secara berkala mengirim catatan heartbeat ke topik Kafka. Catatan heartbeat ini memiliki bidang op dengan nilai MHEARTBEAT. Topik ini menjelaskan format pesan secara keseluruhan, format pesan heartbeat, dan format pesan untuk perubahan data sumber. Untuk informasi lebih lanjut, lihat Tipe bidang dan Parameter.

    Format pesan untuk output single-table real-time

    Saat Anda mengonfigurasi destinasi Kafka untuk tugas sinkronisasi single-table real-time, Anda harus menentukan format nilai untuk catatan yang ditulis ke Kafka. Format yang didukung mencakup Canal CDC dan JSON. Untuk detail lebih lanjut, lihat Lampiran: Deskripsi format output.

    Tipe bidang

    Sistem membaca data dari sumber, memetakannya ke salah satu dari enam tipe (BOOLEAN, DOUBLE, DATE, BYTES, LONG, atau STRING), dan menuliskannya ke topik Kafka dalam format JSON.

    Tipe

    Deskripsi

    BOOLEAN

    Bersesuaian dengan tipe boolean JSON. Nilai yang valid adalah true dan false.

    DATE

    Bersesuaian dengan tipe angka JSON. Nilainya berupa timestamp Unix 13 digit dengan presisi milidetik (ms).

    BYTES

    Bersesuaian dengan tipe string dalam JSON. Sebelum ditulis ke Kafka, array byte dikodekan Base64 menjadi string. Saat dikonsumsi, data tersebut harus didekode Base64 (Encoding: Base64.getEncoder().encodeToString(text.getBytes("UTF-8")). Decoding: Base64.getDecoder().decode(encodedText)).

    STRING

    Bersesuaian dengan tipe string JSON.

    LONG

    Bersesuaian dengan tipe angka JSON.

    DOUBLE

    Bersesuaian dengan tipe angka JSON.

    Parameter

    Tabel berikut menjelaskan setiap bidang dalam pesan yang ditulis ke Kafka.

    Bidang level-1

    Bidang level-2

    Deskripsi

    schema

    dataColumn

    Tipe datanya adalah JSONArray. Parameter dataColumn mencatat informasi tipe untuk semua kolom dari catatan perubahan data hulu. Operasi perubahan mencakup perubahan data di database, seperti penambahan, penghapusan, dan pembaruan, serta perubahan pada skema tabel database.

    • name: Nama kolom.

    • type: Tipe kolom.

    primaryKey

    Daftar nama kolom kunci primer.

    source

    Objek yang berisi informasi tentang database atau tabel sumber.

    • dbType: Tipe database. String.

    • dbVersion: Versi database. String.

    • dbName: Nama database. String.

    • schemaName: Nama skema, digunakan untuk database seperti PostgreSQL dan SQL Server. String.

    • tableName: Nama tabel. String.

    payload

    before

    JSONObject yang merepresentasikan catatan data sebelum perubahan, juga dikenal sebagai "before image". Misalnya, jika catatan di sumber data MySQL diperbarui, bidang before menyimpan konten catatan sebelum pembaruan tersebut.

    • Bidang ini diisi untuk operasi pembaruan dan penghapusan.

    • dataColumn: Informasi data, bertipe JSONObject. Formatnya adalah Nama kolom:Nilai kolom, di mana nama kolom berupa string dan nilai kolom dapat berupa BOOLEAN, DOUBLE, DATE, BYTES, LONG, atau STRING.

    after

    Data setelah perubahan, juga dikenal sebagai "after image". Formatnya sama dengan bidang before.

    sequenceId

    String unik yang dihasilkan oleh StreamX untuk setiap catatan. Digunakan untuk mengurutkan data saat menggabungkan data penuh dan data inkremental.

    Catatan

    Operasi pembaruan dari sumber menghasilkan dua catatan: catatan "update before" dan catatan "update after". Kedua catatan tersebut memiliki sequenceId yang sama.

    scn

    System Change Number (SCN) dari sumber. Bidang ini hanya berlaku untuk sumber Oracle.

    op

    Jenis operasi yang ditangkap dari sumber. Nilai yang valid:

    • INSERT: Penyisipan data.

    • UPDATE_BEFOR: Before image dari pembaruan.

    • UPDATE_AFTER: After image dari pembaruan.

    • DELETE: Penghapusan data.

    • TRANSACTION_BEGIN: Awal transaksi database.

    • TRANSACTION_END: Akhir transaksi database.

    • CREATE: Pembuatan tabel.

    • ALTER: Perubahan tabel.

    • QUERY: Pernyataan SQL asli untuk perubahan database.

    • TRUNCATE: Pemotongan tabel.

    • RENAME: Operasi penggantian nama tabel.

    • CINDEX: Pembuatan indeks.

    • DINDEX: Penghapusan indeks.

    • MHEARTBEAT: Pesan heartbeat yang menunjukkan tugas sinkronisasi aktif, meskipun sumber tidak memiliki data baru.

    timestamp

    JSONObject yang berisi timestamp terkait catatan data.

    • eventTime: Waktu terjadinya perubahan di database sumber. Timestamp 13 digit dengan presisi milidetik. Long.

    • systemTime: Waktu saat tugas sinkronisasi memproses pesan perubahan tersebut. Timestamp 13 digit dengan presisi milidetik. Long.

    • checkpointTime: Timestamp yang digunakan saat mengatur ulang offset sinkronisasi. Nilai ini biasanya sama dengan eventTime. Timestamp 13 digit dengan presisi milidetik. Long.

    ddl

    Bidang ini diisi dengan data hanya ketika skema tabel database diubah. Untuk perubahan data, seperti penambahan, penghapusan, dan modifikasi, ddl yang bersesuaian langsung diatur ke null.

    • text: Teks pernyataan DDL. String.

    • ddlMeta: String yang dikodekan Base64 dari objek Java serialisasi yang mencatat perubahan DDL. String.

    version

    N/A

    Versi format pesan.

    Format pesan secara keseluruhan

    Kode berikut menunjukkan format umum pesan Kafka:

    {
    "schema": { // Metadata tentang perubahan, termasuk nama dan tipe kolom.
        "dataColumn": [// Informasi tentang kolom data yang berubah, digunakan untuk memperbarui catatan di destinasi.
            {
                "name": "id",
                "type": "LONG"
            },
            {
                "name": "name",
                "type": "STRING"
            },
            {
                "name": "binData",
                "type": "BYTES"
            },
            {
                "name": "ts",
                "type": "DATE"
            },
            {
              "name":"rowid",// Jika sumber data adalah Oracle, rowid disertakan dalam kolom data.
              "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"// Tipe string. Rowid dari sumber Oracle.
            }
        },
        "after": {
            "dataColumn":{
                "id": 222,
                "name":"donald",
                "binData": "[base64 string]",
                "ts": 1590315269000,
                "rowid": "AAIUMPAAFAACxExAAE"// Tipe string. Rowid dari sumber Oracle.
            }
        },
        "sequenceId":"XXX",// Tipe string. Digunakan untuk mengurutkan data saat menggabungkan data penuh dan inkremental.
        "scn":"xxxx",// Tipe string. System Change Number (SCN) dari sumber Oracle.
        "op": "INSERT/UPDATE_BEFOR/UPDATE_AFTER/DELETE/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/MHEARTBEAT...",// Nilainya case-sensitive.
        "timestamp": {
            "eventTime": 1,// Wajib. Waktu event perubahan di database sumber. Timestamp 13 digit dengan presisi milidetik.
            "systemTime": 2,// Opsional. Waktu saat tugas sinkronisasi memproses pesan perubahan ini. Timestamp 13 digit dengan presisi milidetik.
            "checkpointTime": 3// Opsional. Timestamp yang digunakan saat mengatur ulang offset sinkronisasi. Biasanya sama dengan eventTime.
        },
        "ddl": {
            "text": "ADD COLUMN ...",
            "ddlMeta": "[String Base64-encoded dari objek SQLStatement yang diserialisasi]"
        }
    },
    "version":"1.0.0"
}
    Catatan

    Untuk informasi lebih lanjut tentang tipe bidang dan parameter, lihat Tipe bidang dan Parameter.

    Format pesan heartbeat

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

    Untuk informasi lebih lanjut tentang tipe bidang dan parameter, lihat Tipe bidang dan Parameter.

    Format pesan untuk perubahan data sumber

    • Format pesan untuk operasi insert:

      {
    "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 pesan untuk operasi update:

      • Jika opsi "When one record in the source is updated, one Kafka record is generated." tidak dipilih, operasi update menghasilkan dua pesan Kafka: satu untuk status data sebelum update ("before image") dan satu untuk status setelah update ("after image"). Contoh kode berikut menunjukkan formatnya:

        Format pesan untuk before image:

        {
    "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 pesan untuk after image:

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

      • Jika opsi "When one record in the source is updated, one Kafka record is generated." dipilih, operasi update menghasilkan satu pesan Kafka yang berisi both before dan after image data. Contoh kode berikut menunjukkan formatnya:

        {
    "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 pesan untuk operasi delete:

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

    Untuk informasi lebih lanjut tentang tipe bidang dan parameter, lihat Tipe bidang dan Parameter.