Format of messages written to Kafka - DataWorks - Alibaba Cloud Dokümantasyon Merkezi

Arka plan

Tam veritabanı senkronizasyon görevi, yukarı akış kaynağından veri okur ve aşağıda açıklanan JSON biçiminde bir Kafka konusuna yazar. Bir mesaj, değişiklik kaydına ait sütun bilgilerini ve değişiklik öncesi ile sonrası veri durumlarını içerir. Kafka tüketicilerinin görev ilerlemesini belirleyebilmesi için görev, op alanı MHEARTBEAT olarak ayarlanmış bir kalp atışı kaydını da düzenli aralıklarla konuya yazar. Aşağıdaki bölümlerde Genel mesaj biçimi, Senkronizasyon görevi kalp atışı mesaj biçimi ve Kaynak veri değişiklikleri için mesaj biçimi açıklanmaktadır. Alan türleri ve parametre açıklamaları için bkz. Alan türleri ve Parametre açıklamaları.

Tek tablo gerçek zamanlı çıktı mesaj biçimi

Tek tablo gerçek zamanlı görevler Kafka hedefini yapılandırırken Kafka kayıtlarındaki Value alanının biçimini belirtin. Canal CDC ve JSON desteklenir. Ayrıntılar için bkz. Ek: Çıktı biçimlerinin açıklaması.

Alan türleri

Kaynaktan okunan veriler, geçerli JSON biçiminde Kafka konusuna yazılmadan önce BOOLEAN, DOUBLE, DATE, BYTES, LONG ve STRING olmak üzere altı türden birine eşlenir.

Tür	Açıklama
BOOLEAN	JSON Boolean türüne eşlenir; değerler true ve false'tur.
DATE	JSON sayı türüne eşlenir; milisaniye hassasiyetinde 13 basamaklı bir zaman damgasıdır.
BYTES	JSON dize türüne eşlenir. Bayt dizileri Kafka'ya yazılmadan önce Base64 ile kodlanır ve tüketiciler tarafından Base64 çözülerek işlenmelidir (kodlama: Base64.getEncoder().encodeToString(text.getBytes("UTF-8")); çözümleme: Base64.getDecoder().decode(encodedText)）.
STRING	JSON dize türüne eşlenir.
LONG	JSON sayı türüne eşlenir.
DOUBLE	JSON sayı türüne eşlenir.

Parametre açıklamaları

Aşağıdaki tabloda Kafka'ya yazılan mesajlardaki alanlar açıklanmaktadır.

Üst düzey öğe	İkinci düzey öğe	Açıklama
schema	dataColumn	Veri sütunu türü bilgilerini içeren JSONArray. dataColumn, yukarı akıştaki değişiklik kayıtlarındaki tüm sütunları ve türlerini kaydeder. Değişiklikler arasında veri ekleme, silme veya güncelleme ve veritabanı tablo şemalarındaki değişiklikler yer alır. name: sütun adı type: sütun türü
	primaryKey	Birincil anahtar bilgilerini içeren liste. pk: birincil anahtar adı.
	source	Kaynak veritabanı veya tablo bilgilerini içeren nesne. dbType: String, veritabanı türü. dbVersion: String, veritabanı sürümü. dbName: String, veritabanı adı. schemaName: String, şema adı (Postgres, SQL Server ve benzeri veritabanları için). tableName: String, tablo adı.
payload	before	Bir güncelleme öncesindeki verileri içeren JSONObject. Örneğin, bir MySQL kaynağı bir kaydı güncellediğinde before, güncelleme öncesi kayıt verilerini içerir. Bu alan, kaynaktan okunan güncelleme ve silme mesajları için doldurulur. dataColumn: sütun adı: sütun değeri biçimindeki verileri içeren JSONObject. Sütun adları dizedir; değerler BOOLEAN, DOUBLE, DATE, BYTES, LONG veya STRING olabilir.
	after	Güncelleme sonrası veriler, before ile aynı biçimdedir.
	sequenceId	Streamx tarafından artımlı ve tam verilerin birleştirme sırasında sıralanması için oluşturulan dize. Her Streamx kaydının benzersiz bir sequenceId'si vardır. Not Kaynaktan okunan bir güncelleme mesajı, aynı sequenceId'ye sahip update before ve update after olmak üzere iki çıktı kaydı üretir.
	scn	Kaynak Oracle olduğunda geçerlidir; Oracle SCN'yi içerir.
	op	Kaynaktan okunan mesaj türü. Geçerli değerler: INSERT: veri eklendi. UPDATE_BEFOR: veri güncellemesinden önce. UPDATE_AFTER: veri güncellemesinden sonra. DELETE: veri silindi. TRANSACTION_BEGIN: veritabanı işlemi başlatıldı. TRANSACTION_END: veritabanı işlemi sonlandırıldı. CREATE: veritabanı tablosu oluşturuldu. ALTER: veritabanı tablosu değiştirildi. QUERY: veritabanı değişikliğine ilişkin özgün SQL. TRUNCATE: veritabanı tablosu temizlendi. RENAME: veritabanı tablosu yeniden adlandırıldı. CINDEX: dizin oluşturuldu. DINDEX: dizin silindi. MHEARTBEAT: yeni kaynak verisi olmadığında senkronizasyonun normal şekilde devam ettiğini belirten kalp atışı.
	timestamp	Bu kayıtla ilgili zaman damgalarını içeren JSONObject. eventTime: Long, kaynak veritabanındaki değişikliğin gerçekleştiği zaman, 13 basamaklı milisaniye zaman damgası olarak. systemTime: Long, görevin bu değişiklik mesajını işlediği zaman, 13 basamaklı milisaniye zaman damgası olarak. checkpointTime: Long, senkronizasyon kontrol noktası sıfırlandığında ayarlanan zaman, 13 basamaklı milisaniye zaman damgası olarak. Genellikle eventTime'a eşittir.
	ddl	Bu alan yalnızca veritabanı tablo şema değişiklikleri için doldurulur. Ekleme, silme ve güncelleme dahil veri değişiklikleri için ddl alanı null olarak ayarlanır. text: String, veritabanı DDL ifadesi. ddlMeta: String, veritabanı DDL değişiklik türünü kaydeden bir Java nesnesinin serileştirilmesiyle üretilen Base64 kodlu dize.
version	Yok	Mesaj biçimi sürümü.

Genel mesaj biçimi

Kafka'ya yazılan bir mesajın genel biçimi şu şekildedir:

{ "schema": { //Değişiklik meta verileri, yalnızca sütun adlarını ve türlerini içerir "dataColumn": [//Hedef tablo kaydını güncellemek için değiştirilen veri sütunu bilgileri { "name": "id", "type": "LONG" }, { "name": "name", "type": "STRING" }, { "name": "binData", "type": "BYTES" }, { "name": "ts", "type": "DATE" }, { "name":"rowid",// Oracle kaynağı için rowid, veri sütunları arasında yer alır "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"//Oracle rowid'yi içeren dize } }, "after": { "dataColumn":{ "id": 222, "name":"donald", "binData": "[base64 string]", "ts": 1590315269000, "rowid": "AAIUMPAAFAACxExAAE"//Oracle rowid'yi içeren dize } }, "sequenceId":"XXX",//Artımlı ve tam verileri birleştirirken verileri sıralamak için kullanılan dize, "scn":"xxxx",//Oracle SCN'yi içeren dize "op": "INSERT/UPDATE_BEFOR/UPDATE_AFTER/DELETE/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/MHEARTBEAT...",//Büyük/küçük harfe duyarlı, "timestamp": { "eventTime": 1,//Gerekli: kaynak veritabanındaki değişiklik zamanı, 13 basamaklı milisaniye zaman damgası "systemTime": 2,//İsteğe bağlı: görevin değişikliği işlediği zaman, 13 basamaklı milisaniye zaman damgası "checkpointTime": 3//İsteğe bağlı: senkronizasyon kontrol noktası sıfırlanırken ayarlanan zaman, 13 basamaklı milisaniye zaman damgası, genellikle eventTime'a eşittir }, "ddl": { "text": "ADD COLUMN ...", "ddlMeta": "[SQLStatement serileştirilmiş ikili veri, base64 dizesi olarak ifade edilir]" } }, "version":"1.0.0" }

Not

Alan türleri ve parametre açıklamaları için bkz. Alan türleri ve Parametre açıklamaları.

Senkronizasyon görevi kalp atışı mesaj biçimi

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

Not

Alan türleri ve parametre açıklamaları için bkz. Alan türleri ve Parametre açıklamaları.

Kaynak veri değişiklikleri için mesaj biçimi

Kaynak ekleme için Kafka mesaj biçimi:

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

Kaynak güncelleme için Kafka mesaj biçimi:

When one record in the source is updated, one Kafka record is generated. seçili değilken, kaynak güncellemesi iki Kafka mesajı üretir; biri güncelleme öncesi durum, diğeri güncelleme sonrası durumu içindir. Biçimler şöyledir:

Güncelleme öncesi durum için mesaj biçimi:

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

Güncelleme sonrası durum için mesaj biçimi:

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

When one record in the source is updated, one Kafka record is generated. seçili iken, kaynak güncellemesi hem önceki hem de sonraki durumları içeren tek bir Kafka mesajı üretir. Biçim şöyledir:

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

Kaynak silme için Kafka mesaj biçimi:

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

Not

Alan türleri ve parametre açıklamaları için bkz. Alan türleri ve Parametre açıklamaları.