メッセージキューのデータストレージフォーマット - Data Transmission Service

データストレージフォーマット

DTS は、メッセージキューに書き込まれるデータに対して、次の 4 つのストレージフォーマットをサポートします。

DTS Avro：データ構造やオブジェクトを保存または転送しやすいフォーマットに変換するデータシリアル化フォーマットです。
Shareplex Json：データレプリケーションソフトウェア SharePlex がソースデータベースからデータを読み取り、メッセージキューにデータを書き込む際、データは Shareplex Json フォーマットで保存されます。
Canal Json：Canal はデータベースから増分ログを解析し、増分データをメッセージキューに送信します。データは Canal Json フォーマットで保存されます。

DTS Avro

DTS Avro のスキーマ定義に基づいてデータを解析する必要があります。詳細については、「DTS Avro のスキーマ定義」および「DTS Avro のデシリアライズ例」をご参照ください。

説明

DTS Avro フォーマットでは、DDL ステートメントは String 型です。

Shareplex Json

パラメータの説明

パラメータ	説明
`time`	データベースでトランザクションがコミットされた時刻です。フォーマットは yyyy-MM-ddTHH:mm:ssZ (UTC) です。
`userid`	トランザクションをコミットしたユーザーの ID です。
`op`	データ操作タイプです。有効な値には、INSERT、UPDATE、DELETE、TRUNCATE、DROP COLUMN、UPDATE BEFORE、UPDATE AFTER が含まれます。
`scn`	システム変更番号 (SCN) です。データベースで特定の時刻にコミットされたトランザクションのバージョンを識別します。コミットされた各トランザクションには、一意の SCN が割り当てられます。
`rowid`	データベース内のレコードを特定するために使用される、相対的に一意なアドレス値です。
`trans`	トランザクション ID です。
`seq`	トランザクション内の操作の序数です。値は 1 から始まります。
`size`	トランザクション内の操作の合計数です。
`table`	テーブル名です。
`idx`	トランザクション内の操作のインデックスです。形式は `seq/size` です。たとえば、`1/11` は、11 個の操作を含むトランザクション内の最初の操作であることを示します。
`posttime`	宛先データベースにトランザクションがコミットされた時刻です。

例

データの挿入

{
    "meta": {
        "time": "2017-06-16T14:24:34", 
        "userid": 84,                                    
        "op": "ins",                                   
          "scn": "14589063118712",                  
          "rowid": "AAATGpAAIAAItcIAAA",      
        "trans": "7.0.411499",                 
        "seq": 1,                                          
        "size": 11,                                         
        "table": "CL_BIZ1.MIO_LOG",       
          "idx": "1/11",                                       
        "posttime": "2017-06-16T14:33:52"
    },
    "data": {
        "MIO_LOG_ID": "32539737"
     }
}

データの更新

{
    "meta": {
        "time": "2017-06-16T15:38:13",
        "userid": 84,
        "op": "upd",                             
        "table": "CL_BIZ1.MIO_LOG"
        ….
    },
    "data": {                                          
        "CNTR_NO": "1171201606"
    },
    "key": {                                            
        "MIO_LOG_ID": "32537893",
        "PLNMIO_REC_ID": "31557806",
        "POL_CODE": null,
        "CNTR_TYPE": null,
        "CNTR_NO": "1171201606syui26"
    }
}

データの削除

{
    "meta": {
        "time": "2017-06-16T15:51:35",
        "userid": 84,
        "op": "del",                      
     },
    "data": {                                    
        "MIO_LOG_ID": "32539739",
        "PLNMIO_REC_ID": "31557806",
        "POL_CODE": null,
        "CNTR_TYPE": null,
        "CG_NO": null
     }
}

Canal Json

説明

[パーティションキー更新後のメッセージ配信の分割] を有効にすると、パーティションキーが変更されたときに Kafka は DELETE メッセージと INSERT メッセージを配信します。 Kafka は、それぞれのパーティションキーの値に基づいて、各メッセージのパーティションを選択します。

例: パーティションキー id の値は 1 です。メッセージは partition-1 に配信されます。次のリストでは、[パーティションキー更新後のメッセージ配信の分割] を有効にする前と後の違いについて説明します。

有効：ソースデータベースで UPDATE SET id = 2 WHERE id = 1 コマンドを実行すると、id=1 の DELETE メッセージが partition-1 に配信され、id=2 の INSERT メッセージが partition-2 に配信されます。
無効：1 つの UPDATE メッセージのみが partition-1 に配信されます。UPDATE 操作は、変更前の値に基づいて配信のパーティションを選択します。

フィールドの説明

パラメータ	説明
`database`	データベース名です。
`es`	ソースデータベースで操作が実行された時刻です。13 桁の UNIX タイムスタンプ (ミリ秒単位) です。説明実際の実行時刻は秒単位です。単位をミリ秒に変換するために、3 つのゼロが追加されます。検索エンジンを使用して UNIX タイムスタンプ変換ツールを見つけることができます。
`id`	操作のシリアル番号です。説明これはタイムスタンプと DTS 内部のオフセットから生成されます。レコードの順序を判断するのに役立ちます。
`isDdl`	操作が DDL 操作であるかどうかを示します。 true：はい。 false：いいえ。
`mysqlType`	フィールドのデータタイプです。説明精度などのデータタイプのパラメータはサポートされていません。
`old` および `data`	変更前または変更後のデータです。説明 2022年3月20日より前に作成された同期または移行インスタンスでは、old の値が変更後のデータを、`data` の値が変更前のデータをそれぞれ示します。デフォルトでは、`old` の値には、変更された列だけでなく、すべての列が含まれます。オープンソースコミュニティに合わせるため、2022年3月20日以降に作成または再起動されたインスタンスでは、`data` の値が変更後のデータを、`old` の値が変更前のデータをそれぞれ示します。
`pkNames`	プライマリキー名です。
`sql`	SQL ステートメントです。
`sqlType`	変換されたフィールドタイプです。値は dataTypeNumber の値と同じです。詳細については、「フィールドタイプと dataTypeNumber 値のマッピング」をご参照ください。
`table`	テーブル名です。
`ts`	宛先データベースへのデータ書き込みが開始された時刻です。13 桁の UNIX タイムスタンプ (ミリ秒単位) です。説明検索エンジンを使用して UNIX タイムスタンプ変換ツールを見つけることができます。
`type`	DELETE、UPDATE、INSERT などの操作タイプです。説明完全データ同期または移行タスクでは、値は [INIT] に固定されます。
`gtid`	グローバルトランザクション識別子 (GTID) です。GTID はグローバルに一意です。各トランザクションは 1 つの GTID に対応します。説明 DTS は GTID フィールドの同期をサポートしていません。このフィールドの値はデフォルトで NULL です。

例

データの削除

説明

2022 年 3 月 20 日より前に作成された同期または移行インスタンスの場合、ソーステーブルからの DELETE ステートメントが Kafka に同期または移行されると、old フィールドにデータが含まれ、data フィールドは null になります。オープンソースコミュニティに合わせるため、2022 年 3 月 20 日以降に作成または再起動されたインスタンスでは、data フィールドにデータが含まれ、old フィールドは null になります。

2022 年 3 月 20 日より前に作成された同期インスタンスまたは移行インスタンス

{
    "old": [
        {
            "shipping_type": "aaa"
        }
    ], 
    "database": "dbname", 
    "es": 1600161894000, 
    "id": 58, 
    "isDdl": false, 
    "mysqlType": {
        "id": "bigint", 
        "shipping_type": "varchar"
    }, 
    "pkNames": [
        "id"
    ], 
    "sql": "", 
    "sqlType": {
        "id": -5, 
        "shipping_type": 12
    }, 
    "table": "tablename", 
    "ts": 1600161894771, 
    "type": "DELETE"
}

2022 年 3 月 20 日以降に作成または再起動された同期インスタンスまたは移行インスタンス

{
    "data": [
        {
            "id": "500000287", 
            "shipping_type": null
        }
    ], 
    "database": "dbname", 
    "es": 1600161894000, 
    "id": 58, 
    "isDdl": false, 
    "mysqlType": {
        "id": "bigint", 
        "shipping_type": "varchar"
    }, 
    "pkNames": [
        "id"
    ], 
    "sql": "", 
    "sqlType": {
        "id": -5, 
        "shipping_type": 12
    }, 
    "table": "tablename", 
    "ts": 1600161894771, 
    "type": "DELETE"
}

DDL 操作

{
    "database":"dbname", // 同期または移行のデータベース名
    "es":1600161894000, // ソースデータがバイナリログに書き込まれた時刻
    "id":58, // DTS キャッシュ内のオフセット
    "isDdl":true, // DDL ステートメントを同期または移行するかどうか
    "sql":"eg:createxxx", // バイナリログからの DDL ステートメント
    "table":"tablename", // 同期または移行のテーブル名
    "ts":1600161894771, // DTS が宛先にデータを書き込んだ時刻
    "type":"DDL"
}