JSON メッセージ形式 - イベントストリーミングの変更

適用対象:SQL Server 2025 (17.x)Azure SQL Database Azure SQL マネージドインスタンス Microsoft Fabric の SQL データベース

この記事では、SQL Server 2025(17.x)、Azure SQL Database、Azure SQL Managed InstanceのChange Event Streaming(CES)機能を使用した際に、Change Event Streaming(CES)機能を使用した際にAzure Event HubsまたはFabric EventstreamにストリーミングされるCloudEventsメッセージフォーマットについて説明しますおよびMicrosoft FabricのSQLデータベース。

注

変更イベントのストリーミングは現在プレビュー段階で、製品間でサポート性に違いがあります。プレビュー期間中、この機能は変更される可能性があります。

概要

Change Event Streamingは CloudEvents 仕様に準拠したイベントを発行するため、イベント駆動型システムと簡単に統合できます。すべての CES CloudEvent には、11 個の属性 (フィールド) が含まれています。 CESはCloudEvent全体、 data 属性を含めてネイティブJSONまたはAvroバイナリとしてシリアライズすることができます。ネイティブのJSONイベントにはAvroのバイナリセクションは含まれていません。両方のシリアライズ形式において、 data 属性はバイト配列型を持ちます。バイトは選択したシリアライズ形式に応じたJSONまたはAvroバイナリエンコーディングを使用し、 CESのデータ属性Avroスキーマに従います。

Important

2026年8月15日現在、AMQPプロトコルはチェンジイベントストリーミング(CES)のために非推奨となっています。プラットフォームごとに違いがあります。移行手順やタイムラインについては、 AMQPプロトコルの廃止を参照してください。

該当する場合、このセクションの説明は CloudEvent仕様から来ており、さらに詳細が含まれています。

属性

specversion:
- データ型: 文字列
- 必要な CloudEvent 属性
- イベントが使用する CloudEvents 仕様のバージョン。このバージョンは文脈の解釈を可能にします。
type
- データ型: 文字列
- 必要な CloudEvent 属性
- 発生元の発生に関連するイベントの種類を表す値を格納します。この値の形式は生産者によって定義され、型のバージョンなどの情報を含む場合があります。詳細については、 CloudEventsのバージョン管理をご覧ください。
- 変更イベントストリーミングイベントのタイプは現在com.microsoft.SQL.CES.DML.V{n}であり、{n}は変更イベントストリーミングDMLイベントスキーマのバージョンMicrosoftを示します。
  - 現在の最新のスキーマバージョンは1です。
source
- データ型: 文字列
- 必要な CloudEvent 属性
- イベントが発生したコンテキストを識別します。ソースとIDの組み合わせは、各イベントごとに一意でなければなりません。現在、このフィールドはSQLからストリーミングされるイベントで常に \/ として送信されています。
id
- データ型: 文字列
- 必要な CloudEvent 属性
- イベントを識別します。プロデューサーは、ソースとIDの組み合わせが各異なるイベントごとに一意であることを保証しなければなりません。重複するイベントが再送信された場合 (たとえば、ネットワークエラーが原因)、同じ ID を持つことができます。コンシューマーは、ソースと ID が同一のイベントが重複していると見なす場合があります。
logicalid
- データ型: 文字列
- 拡張属性
- 共有論理IDは分割されたメッセージを識別します(イベントハブのメッセージサイズ制限によるものです)。
time
- データ型: タイムスタンプ
- オプションの CloudEvent 属性
- コミットが最初にストリームイベントをトリガーしたSQLトランザクション内で発生したUTCタイムスタンプ。
datacontenttype
- データ型: 文字列
- オプションの CloudEvent 属性
- データ値のコンテンツタイプ。この属性を使用すると、データは任意の種類のコンテンツを保持できます。形式とエンコードは、選択したイベント形式とは異なる場合があります。たとえば、JSON エンベロープ形式を使用してレンダリングされたイベントでは、データに XML ペイロードが含まれる可能性があり、コンシューマーには、この属性が "application/xml" に設定されたことが通知されます。異なる datacontenttype 値に対するデータコンテンツのレンダリングルールはイベントフォーマット仕様で定義されています。
operation
- データ型: 文字列
- 拡張属性
- 発生したSQL操作の種類を表しています:
  - インサート用INS
  - アップデート
  - 削除のためのDELです
segmentindex
- データ型: 整数
- 拡張属性
- セグメントインデックスは、論理的なメッセージチャンク内でのメッセージの位置を示します。セグメントインデックスは、メッセージが論理メッセージフラグメントのシーケンス内のどこに存在するかについての情報を提供します。この場は常に存在します。 logicalid、segmentindex、finalsegmentフィールドを使って、構成されたmax_message_size_kb値に基づいて大きなSQLペイロードの分散を表すイベントをソートします。
finalsegment
- データ型: Boolean
- 拡張属性
- このセグメントがシーケンスの最終セグメントかどうかを示します。このフィールドは常に存在し、SQLイベントが設定された max_message_size_kb 値に従ってサブイベントに分割されたかどうかを識別するのに役立ちます。
data
- データ型:バイト配列
- オプションの CloudEvent 属性
- ドメイン固有のイベントデータを含み、変更を記述します。選択したシリアライズ形式に従って、バイトをJSONまたはAvroバイナリとしてデシリアライズします。非直列化されたデータは CESのデータ属性Avroスキーマに従います。フィールドの詳細については、データ属性フォーマットを参照してください。

注

メッセージ分割は列値切り落としとは別のものです。 CESが data 属性をシリアライズする前に、1MBを超える各ストリームカラムの値を1MBに切り詰めます。 CESは max_message_size_kbに応じて形成されたイベントをメッセージチャンクに分割します。

例示

JSON メッセージの例 - 挿入

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "56cb8ff3-5c55-4f3b-a7f7-b044d1933ef6",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000008A80007:00000000000000000001",
  "time": "2026-08-07T16:25:00.890Z",
  "datacontenttype": "application\/json",
  "operation": "INS",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:000008A8:0007\",\"beginlsn\":\"000000B1:000008A8:0003\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:25:00.890Z\"}},\"eventrow\":{\"old\":\"{}\",\"current\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\"}}"
}

JSONメッセージの例 - 更新

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "19221db1-a1b5-4ec7-8937-3fdf9d762abb",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000009300009:00000000000000000001",
  "time": "2026-08-07T16:30:10.123Z",
  "datacontenttype": "application\/json",
  "operation": "UPD",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:00000930:0009\",\"beginlsn\":\"000000B1:00000930:0002\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:30:10.123Z\"}},\"eventrow\":{\"old\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\",\"current\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic-Smith\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\"}}"
}

JSON メッセージの例 - 削除

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "520f9a65-43d7-47f2-94f5-7ea14df635ed",
  "logicalid": "1bf2756a-c15f-4d2e-a2d5-7d3f9dbf85b0:000000B1000009700008:00000000000000000001",
  "time": "2026-08-07T16:35:42.450Z",
  "datacontenttype": "application\/json",
  "operation": "DEL",
  "segmentindex": 0,
  "finalsegment": true,
  "data": "{\"eventsource\":{\"db\":\"EmployeesDb\",\"schema\":\"dbo\",\"tbl\":\"Employees\",\"cols\":[{\"name\":\"Id\",\"type\":\"int\",\"index\":0},{\"name\":\"FirstName\",\"type\":\"nvarchar(50)\",\"index\":1},{\"name\":\"LastName\",\"type\":\"nvarchar(50)\",\"index\":2},{\"name\":\"SignupDate\",\"type\":\"datetime2(7)\",\"index\":3}],\"pkkey\":[{\"columnname\":\"Id\",\"value\":\"8\"}],\"transaction\":{\"commitlsn\":\"000000B1:00000970:0008\",\"beginlsn\":\"000000B1:00000970:0003\",\"sequencenumber\":1,\"finalevent\":false,\"committime\":\"2026-08-07T16:35:42.450Z\"}},\"eventrow\":{\"old\":\"{\\\"Id\\\":\\\"8\\\",\\\"FirstName\\\":\\\"Nikola\\\",\\\"LastName\\\":\\\"Nikolic-Smith\\\",\\\"SignupDate\\\":\\\"2026-08-07 16:25:00.8833333\\\"}\",\"current\":\"{}\"}}"
}

データ属性の形式

data属性はバイト配列です。選択したシリアライズ形式に従って、バイトをJSONまたはAvroバイナリとしてデシリアライズします。どちらの形式でも、得られる Data レコードは CESのデータ属性Avro schema に従い、2つの属性を含みます。

eventsource
eventrow

{
  "data": "{\"eventsource\": {}, \"eventrow\": {\"old\": \"{}\", \"current\": \"{}\"}}"
}

以下のセクションでは、デシリアライズされた属性についてより詳しく説明します。

eventsource

イベントが発生したデータベースとテーブルに関するメタデータについて説明します。

db
- データ型: 文字列
- 説明: テーブルが配置されているデータベースの名前。
- 例: EmployeesDb
schema
- データ型: 文字列
- 説明: テーブルを含むデータベーススキーマ。
- 例: dbo
tbl
- データ型: 文字列
- 説明: イベントが発生したテーブル。
- 例: Employees
cols
- データ型: 配列
- 説明: テーブル内の列の詳細を示す配列。
  - name (文字列):列の名前。
  - type (文字列):列のSQLデータ型で、該当する場合には長さ、精度、またはスケールも含まれます。たとえば、 int、 nvarchar(50)、 datetime2(7)などがあります。
  - index (整数):テーブル内の列のインデックスまたは位置。
pkkey
- データ型: 配列
- 説明: 特定の行を識別するための主キー列とその値を表します。
  - columnname (文字列):主キーで使われる列の名前。
  - value (文字列):主キーで使われる列の値。この値は行を一意に識別するのに役立ちます。
transaction
- データ型:オブジェクト
- 説明:データ操作を含むSQLトランザクションを記述します。
  - commitlsn (文字列):トランザクションのコミットログシーケンス番号(LSN)。
  - beginlsn (文字列):トランザクションの最初のLSN。
  - sequencenumber (整数):トランザクション内のデータ操作の連続番号。この値を使い、トランザクション内のイベントをソートします。
  - finalevent (ブール)使われていません。この場は常に falseの値を持ちます。
  - committime (文字列):トランザクションがデータベースにコミットされた日時。

注

UTC以外のタイムゾーンで設定されたSQL製品では、committimeフィールドにZの接尾辞が誤って含まれますが、このフィールドは公開データベースのローカル時刻を示しています。データベースがUTCを使う場合、値と接尾辞は一致します。この問題は既知のものであり、今後の機能リリースで修正が待っています。

eventrow

行レベルの変更について説明し、レコード内のフィールドの古い値と現在の値を比較します。

old (文字列でラップされたオブジェクト): イベントの前の行の値を表します。
- 各キーと値のペアは、次の要素で構成されます。
  - <column_name>: (文字列): 列の名前。
  - <column_value>: (string/int/etc):その列の前の値。
current (文字列でラップされたオブジェクト): イベントの後の行の更新された値を表します。
- 古いオブジェクトと同様に、各キーと値のペアは次のように構造化されています。
  - <column_name> (文字列): 列の名前。
  - <column_value> (string/int/etc): その列の新しい値または現在の値。

CES CloudEvent Avro schema

{
  "type": "record",
  "name": "ChangeEvent",
  "fields": [
    {
      "name": "specversion",
      "type": "string"
    },
    {
      "name": "type",
      "type": "string"
    },
    {
      "name": "source",
      "type": "string"
    },
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "logicalid",
      "type": "string"
    },
    {
      "name": "time",
      "type": "string"
    },
    {
      "name": "datacontenttype",
      "type": "string"
    },
    {
      "name": "operation",
      "type": "string"
    },
    {
      "name": "segmentindex",
      "type": "int"
    },
    {
      "name": "finalsegment",
      "type": "boolean"
    },
    {
      "name": "data",
      "type": "bytes"
    }
  ]
}

CES data attribute Avro schema

ネイティブJSONおよびAvro binary CloudEventsで data バイト配列をデシリアライズする際は、以下のスキーマを使用してください。

{
  "name": "Data",
  "type": "record",
  "fields": [
    {
      "name": "eventsource",
      "type": {
        "name": "EventSource",
        "type": "record",
        "fields": [
          {
            "name": "db",
            "type": "string"
          },
          {
            "name": "schema",
            "type": "string"
          },
          {
            "name": "tbl",
            "type": "string"
          },
          {
            "name": "cols",
            "type": {
              "type": "array",
              "items": {
                "name": "Column",
                "type": "record",
                "fields": [
                  {
                    "name": "name",
                    "type": "string"
                  },
                  {
                    "name": "type",
                    "type": "string"
                  },
                  {
                    "name": "index",
                    "type": "int"
                  }
                ]
              }
            }
          },
          {
            "name": "pkkey",
            "type": {
              "type": "array",
              "items": {
                "name": "PkKey",
                "type": "record",
                "fields": [
                  {
                    "name": "columnname",
                    "type": "string"
                  },
                  {
                    "name": "value",
                    "type": "string"
                  }
                ]
              }
            }
          },
          {
            "name": "transaction",
            "type": {
              "name": "Transaction",
              "type": "record",
              "fields": [
                {
                  "name": "commitlsn",
                  "type": "string"
                },
                {
                  "name": "beginlsn",
                  "type": "string"
                },
                {
                  "name": "sequencenumber",
                  "type": "int"
                },
                {
                  "name": "finalevent",
                  "type": "boolean"
                },
                {
                  "name": "committime",
                  "type": "string"
                }
              ]
            }
          }
        ]
      }
    },
    {
      "name": "eventrow",
      "type": {
        "name": "EventRow",
        "type": "record",
        "fields": [
          {
            "name": "old",
            "type": "string"
          },
          {
            "name": "current",
            "type": "string"
          }
        ]
      }
    }
  ]
}

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-14