Teilen über

JSON-Nachrichtenformat – Ändern des Ereignisstreamings

Gilt für: SQL Server 2025 (17.x) Azure SQL-Datenbank

Dieser Artikel beschreibt das JSON-Format einer CloudEvents-Nachricht, die von SQL Server zu Azure Event Hubs gestreamt wird, wenn die Change Event Streaming (CES)- Funktion verwendet wird, die in SQL Server 2025 (17.x) und Azure SQL Database eingeführt wurde.

Hinweis

Das Ändern des Ereignisstreamings befindet sich derzeit in der Vorschau für:

Während der Vorschau kann dieses Feature geändert werden. Informationen zur aktuellen Unterstützung finden Sie unter "Einschränkungen".

Überblick

Ereignisse, die durch Änderungsereignisstreaming ausgegeben werden, folgen der CloudEvents-Spezifikation , sodass sie einfach in ereignisgesteuerte Systeme integriert werden können. Alle CES CloudEvents enthalten 11 Attribute (Felder). CES kann so konfiguriert werden, dass CloudEvents als JSON (native) oder als Avro-Binärdatei serialisiert wird. In den folgenden Abschnitten dieses Artikels wird das Nachrichtenformat ausführlich beschrieben, einschließlich CES CloudEvent-Attributen und Serialisierung.

Falls zutreffend, werden die Beschreibungen in diesem Abschnitt aus der CloudEvent-Spezifikation entnommen, die zusätzliche Details enthält.

Attribute

  • specversion:

    • Datentyp: String
    • Erforderliches CloudEvent-Attribut
    • Die Version der CloudEvents-Spezifikation, die das Ereignis verwendet. Dies ermöglicht die Interpretation des Kontexts.

  • type

    • Datentyp: String
    • Erforderliches CloudEvent-Attribut
    • Enthält einen Wert, der den Typ des Ereignisses beschreibt, das mit dem ursprünglichen Vorkommen verknüpft ist. Das Format wird vom Produzenten definiert und kann Informationen wie die Version des Typs enthalten. Weitere Informationen können Sie unter "Versionsverwaltung von CloudEvents" lesen.
    • Für Change Event Streaming-Ereignisse lautet der Typ derzeit: com.microsoft.SQL.CES.DML.V{n}, wobei {n} die Version des Microsoft Change Event Streaming DML-Ereignisschemas anzeigt.
      • Die aktuell neueste Schema-Version ist 1.

  • source

    • Datentyp: String
    • Erforderliches CloudEvent-Attribut
    • Gibt den Kontext an, in dem ein Ereignis aufgetreten ist. Source + ID muss für jedes Ereignis eindeutig sein. Derzeit wird dieses Feld immer als Ereignisse gesendet, die \/ von SQL gestreamt werden.

  • id

    • Datentyp: String
    • Erforderliches CloudEvent-Attribut
    • Identifiziert das Ereignis. Produzenten müssen sicherstellen, dass die Quell-ID für jedes einzelne Ereignis eindeutig ist. Wenn ein dupliziertes Ereignis erneut auftritt (z. B. aufgrund eines Netzwerkfehlers), könnte es dieselbe ID haben. Consumer können davon ausgehen, dass Ereignisse mit identischer Quelle und ID Duplikate sind.

  • logicalid

    • Datentyp: String
    • Erweiterungsattribut
    • Geteilte Nachrichten (aufgrund von Msg-Größenbeschränkungen für Event Hubs) werden durch freigegebene logische IDs identifiziert.

  • time

    • Datentyp: Zeitstempel
    • Optionales CloudEvent-Attribut
    • UTC-Zeitstempel, wann der Commit innerhalb einer SQL-Transaktion stattfand, die ursprünglich ein gestreamtes Ereignis auslöst.

  • Datentyp

    • Datentyp: String
    • Optionales CloudEvent-Attribut
    • Inhaltstyp des Datenwerts. Mit diesem Attribut können Daten beliebige Inhaltstypen übertragen, wobei sich Format und Codierung von dem des ausgewählten Ereignisformats unterscheiden können. Beispielsweise kann ein ereignis, das mit dem JSON-Umschlagformat gerendert wird, eine XML-Nutzlast in den Daten enthalten, und der Verbraucher wird durch dieses Attribut informiert, das auf "application/xml" festgelegt wird. Die Regeln, wie Dateninhalte für verschiedene datacontenttype Werte dargestellt werden, sind in den Ereignisformatspezifikationen definiert

  • operation

    • Datentyp: String
    • Erweiterung
    • Repräsentiert die Art der SQL-Operation, die stattgefunden hat:
      • INS für Einsätze
      • UPD für Updates
      • DEL für Löschungen

  • segmentindex

    • Datentyp: Ganze Zahl
    • Erweiterungsattribut
    • Segmentindex, der die Position der Nachricht innerhalb der logischen Nachrichtenblöcke angibt. Der Segmentindex enthält Informationen dazu, wo die Nachricht in der Sequenz logischer Nachrichtenfragmente steht. Dieses Feld ist immer vorhanden. Verwenden Sie Logicalid + Segmentindex + Finalsegment-Felder, um die eingehenden Ereignisse, die eine große SQL-Nutzlast repräsentieren, in mehrere Ereignisse zu sortieren.

  • finalsegment

    • Datentyp: Boolescher Wert
    • Erweiterungsattribut
    • Gibt an, ob dieses Segment das endgültige Segment der Sequenz ist. Dieses Feld ist immer vorhanden und hilft zu erkennen, ob ein SQL-Ereignis, das zu groß für die maximale konfigurierte Nachrichtengröße war, in Unterereignisse aufgeteilt wurde.

  • data

    • Datentyp: String
    • Optionales CloudEvent-Attribut
    • Domänenspezifische Ereignisdaten. Für CES sind Daten eine Zeichenfolge, die als JSON analysiert werden kann. Dieser JSON-Code beschreibt, wie Daten geändert wurden. Das Format des Datenattributes befindet sich im Daten-Attributformat.

Beispiele

JSON-Meldungsbeispiel – Einfügen

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "d43f09a6-d13b-4902-86d4-17bdb5edb872",
  "logicalid": "9c8d4ad2-bf54-4f10-a96f-038af496997f:0000002C00000300017C:00000000000000000001",
  "time": "2025-03-14T16:45:20.650Z",
  "datacontenttype": "application\/json",
  "operation": "INS",
  "splitindex": 0,
  "splittotalcnt": 0,
  "data": "{\n  \"eventsource\": {\n    \"db\": \"db1\",\n    \"schema\": \"dbo\",\n    \"tbl\": \"Purchases\",\n    \"cols\": [\n      {\n        \"name\": \"purchase_id\",\n        \"type\": \"int\",\n        \"index\": 0\n      },\n      {\n        \"name\": \"customer_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 1\n      },\n      {\n        \"name\": \"product_id\",\n        \"type\": \"int\",\n        \"index\": 2\n      },\n      {\n        \"name\": \"product_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 3\n      },\n      {\n        \"name\": \"price_per_item\",\n        \"type\": \"int\",\n        \"index\": 4\n      },\n      {\n        \"name\": \"quantity\",\n        \"type\": \"int\",\n        \"index\": 5\n      },\n      {\n        \"name\": \"purchase_date\",\n        \"type\": \"datetime\",\n        \"index\": 6\n      },\n      {\n        \"name\": \"payment_method\",\n        \"type\": \"varchar(50)\",\n        \"index\": 7\n      }\n    ],\n    \"pkkey\": [\n      {\n        \"columnname\": \"purchase_id\",\n        \"value\": \"105\"\n      }\n    ]\n  },\n  \"eventrow\": {\n    \"old\": \"{}\",\n    \"current\": \"{\\\"purchase_id\\\": \\\"105\\\", \\\"customer_name\\\": \\\"Anna Doe\\\", \\\"product_id\\\": \\\"101\\\", \\\"product_name\\\": \\\"Game 2077\\\", \\\"price_per_item\\\": \\\"60\\\", \\\"quantity\\\": \\\"1\\\", \\\"purchase_date\\\": \\\"2025-03-14 16:45:01.000\\\", \\\"payment_method\\\": \\\"Credit Card\\\"}\"\n  }\n}"
}

JSON-Nachrichtenbeispiel – aktualisiert

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "c425575f-00bb-45cf-acec-c55fdc7d08cd",
  "logicalid": "9c8d4ad2-bf54-4f10-a96f-038af496997f:0000002C000003500004:00000000000000000001",
  "time": "2025-03-14T16:49:59.567Z",
  "datacontenttype": "application\/json",
  "operation": "UPD",
  "splitindex": 0,
  "splittotalcnt": 0,
  "data": "{\n  \"eventsource\": {\n    \"db\": \"db1\",\n    \"schema\": \"dbo\",\n    \"tbl\": \"Purchases\",\n    \"cols\": [\n      {\n        \"name\": \"purchase_id\",\n        \"type\": \"int\",\n        \"index\": 0\n      },\n      {\n        \"name\": \"customer_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 1\n      },\n      {\n        \"name\": \"product_id\",\n        \"type\": \"int\",\n        \"index\": 2\n      },\n      {\n        \"name\": \"product_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 3\n      },\n      {\n        \"name\": \"price_per_item\",\n        \"type\": \"int\",\n        \"index\": 4\n      },\n      {\n        \"name\": \"quantity\",\n        \"type\": \"int\",\n        \"index\": 5\n      },\n      {\n        \"name\": \"purchase_date\",\n        \"type\": \"datetime\",\n        \"index\": 6\n      },\n      {\n        \"name\": \"payment_method\",\n        \"type\": \"varchar(50)\",\n        \"index\": 7\n      }\n    ],\n    \"pkkey\": [\n      {\n        \"columnname\": \"purchase_id\",\n        \"value\": \"105\"\n      }\n    ]\n  },\n  \"eventrow\": {\n    \"old\": \"{}\",\n    \"current\": \"{\\\"purchase_id\\\": \\\"105\\\", \\\"customer_name\\\": \\\"Anna Doe\\\", \\\"product_id\\\": \\\"100\\\", \\\"product_name\\\": \\\"Game 2066\\\", \\\"price_per_item\\\": \\\"50\\\", \\\"quantity\\\": \\\"2\\\", \\\"purchase_date\\\": \\\"2025-03-14 16:45:01.000\\\", \\\"payment_method\\\": \\\"Credit Card\\\"}\"\n  }\n}"
}

JSON-Nachrichtenbeispiel – Löschen

{
  "specversion": "1.0",
  "type": "com.microsoft.SQL.CES.DML.V1",
  "source": "\/",
  "id": "24fa0c2c-c45d-4abf-9a8d-fba04c29fc86",
  "logicalid": "9c8d4ad2-bf54-4f10-a96f-038af496997f:0000002C000003600019:00000000000000000001",
  "time": "2025-03-14T16:51:39.613Z",
  "datacontenttype": "application\/json",
  "operation": "DEL",
  "splitindex": 0,
  "splittotalcnt": 0,
  "data": "{\n  \"eventsource\": {\n    \"db\": \"db1\",\n    \"schema\": \"dbo\",\n    \"tbl\": \"Purchases\",\n    \"cols\": [\n      {\n        \"name\": \"purchase_id\",\n        \"type\": \"int\",\n        \"index\": 0\n      },\n      {\n        \"name\": \"customer_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 1\n      },\n      {\n        \"name\": \"product_id\",\n        \"type\": \"int\",\n        \"index\": 2\n      },\n      {\n        \"name\": \"product_name\",\n        \"type\": \"varchar(100)\",\n        \"index\": 3\n      },\n      {\n        \"name\": \"price_per_item\",\n        \"type\": \"int\",\n        \"index\": 4\n      },\n      {\n        \"name\": \"quantity\",\n        \"type\": \"int\",\n        \"index\": 5\n      },\n      {\n        \"name\": \"purchase_date\",\n        \"type\": \"datetime\",\n        \"index\": 6\n      },\n      {\n        \"name\": \"payment_method\",\n        \"type\": \"varchar(50)\",\n        \"index\": 7\n      }\n    ],\n    \"pkkey\": [\n      {\n        \"columnname\": \"purchase_id\",\n        \"value\": \"105\"\n      }\n    ]\n  },\n  \"eventrow\": {\n    \"old\": \"{\\\"purchase_id\\\": \\\"105\\\", \\\"customer_name\\\": \\\"Anna Doe\\\", \\\"product_id\\\": \\\"100\\\", \\\"product_name\\\": \\\"Game 2066\\\", \\\"price_per_item\\\": \\\"50\\\", \\\"quantity\\\": \\\"2\\\", \\\"purchase_date\\\": \\\"2025-03-14 16:45:01.000\\\", \\\"payment_method\\\": \\\"Credit Card\\\"}\",\n    \"current\": \"{}\"\n  }\n}"
}

Datenattributeformat

Daten sind ein JSON-Objekt, das in ein Zeichenfolgen-Attribut eingeschlossen ist, das zwei Attribute enthält:

  • eventSource
  • eventRow
"data": "{ "eventsource": {<eventSource>}, "eventdata": {<eventData>}}"

Details zu diesen beiden Attributen werden in den folgenden Abschnitten ausführlicher erläutert:

eventsource

Beschreibt die Metadaten zu der Datenbank und der Tabelle, in der das Ereignis aufgetreten ist:

  • db

    • Datentyp: String
    • Beschreibung: Der Name der Datenbank, in der sich die Tabelle befindet.
    • Beispiel: cessqldb001

  • schema

    • Datentyp: String
    • Beschreibung: Das Datenbankschema, das die Tabelle enthält.
    • Beispiel: dbo

  • tbl

    • Datentyp: String
    • Beschreibung: Die Tabelle, in der das Ereignis aufgetreten ist.
    • Beispiel: Purchases

  • cols

    • Datentyp: Array
    • Beschreibung: Ein Array, das die Spalten in der Tabelle enthält.
      • Name (Zeichenfolge): Der Name der Spalte.
      • Typ (Zeichenkette): Der Datentyp der Spalte (VARCHAR oder INT).
      • index (ganze Zahl): Der Index oder die Position der Spalte in der Tabelle.

  • pkkey

    • Datentyp: Array
    • Beschreibung: Stellt die Primärschlüsselspalten und deren Werte für die Identifizierung der bestimmten Zeile dar.
      • columnname (string): Der Name der spalte, die im Primärschlüssel verwendet wird.
      • wert (string/int/etc.): Der Wert für die spalte, die im Primärschlüssel verwendet wird, hilft, die Zeile eindeutig zu identifizieren.

eventrow

Beschreibt Änderungen auf Zeilenebene und vergleicht die alten und aktuellen Werte der Felder im Datensatz.

  • altes (in Zeichenfolge eingeschlossenes Objekt): Stellt die Werte in der Zeile vor dem Ereignis dar.
    • Jedes Schlüssel-Wert-Paar besteht aus:
      • <column_name>: (Zeichenfolge): Der Name der Spalte.
      • <column_value>: (string/int/etc.): Der vorherige Wert für diese Spalte.
  • current (objekt wrapped in string): Represents the updated values in the row after the event.
    • Ähnlich wie das alte Objekt, wobei jedes Schlüsselwertpaar wie folgt strukturiert ist:
      • <column_name> (Zeichenfolge): Der Name der Spalte.
      • <column_value> (string/int/etc.): Der neue oder aktuelle Wert für diese Spalte.

CES CloudEvent JSON-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"
    }
  ]
}

JSON-Schema des CES-Datenattributes

{
  "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": "committime",
                  "type": "string"
                }
              ]
            }
          }
        ]
      }
    },
    {
      "name": "eventrow",
      "type": {
        "name": "EventRow",
        "type": "record",
        "fields": [
          {
            "name": "old",
            "type": "string"
          },
          {
            "name": "current",
            "type": "string"
          }
        ]
      }
    }
  ]
}

Verwandte Inhalte

  • Last updated on