Format komunikatów JSON — zmienianie przesyłania strumieniowego zdarzeń

Dotyczy: SQL Server 2025 (17.x) Azure SQL DatabaseAzure SQL Managed InstanceBaza danych SQL w Microsoft Fabric

Ten artykuł opisuje format wiadomości CloudEvents, który przesyła się do Azure Event Hubs lub Fabric Eventstream, gdy korzystasz z funkcji change event streaming (CES) w SQL Server 2025 (17.x), Azure SQL Database, Azure SQL Managed Instance, oraz bazy danych SQL w Microsoft Fabric.

Przegląd

Strumieniowanie zdarzeń zmian generuje zdarzenia zgodne ze specyfikacją CloudEvents , dzięki czemu łatwo można je zintegrować z systemami opartymi na zdarzeniach. Wszystkie elementy CES CloudEvents zawierają 11 atrybutów (pól). Możesz skonfigurować CES tak, aby cały CloudEvent, włącznie z atrybutem data , serializował jako natywny plik binarny JSON lub Avro. Natywne zdarzenia JSON nie zawierają sekcji binarnych Avro. W obu formatach data serializacji atrybut ma typ bajt-array. Bajty wykorzystują kodowanie binarne JSON lub Avro zgodnie z wybranym formatem serializacji i podążają za schematem atrybutu danych CES Avro.

Powiązane specyfikacje i zasoby

Gdy jest to możliwe, opisy w tej sekcji pochodzą ze specyfikacji CloudEvent, która zawiera więcej szczegółów.

Atrybuty

• specversion:

  • Typ danych: Ciąg
  • Wymagany atrybut CloudEvent
  • Wersja specyfikacji CloudEvents używana przez zdarzenie. Ta wersja umożliwia interpretację kontekstu.

• type

  • Typ danych: Ciąg
  • Wymagany atrybut CloudEvent
  • Zawiera wartość opisjącą typ zdarzenia powiązanego z wystąpieniem źródłowym. Format tej wartości jest definiowany przez producenta i może zawierać informacje takie jak wersja typu. Więcej informacji można znaleźć w artykule Wersjonowanie CloudEvents.
  • Dla event event streaming typ jest obecny: com.microsoft.SQL.CES.DML.V{n}, gdzie {n} oznacza wersję schematu zdarzeń DML Microsoft change event streaming.
    • Aktualna najnowsza wersja schematu to 1.

• source

  • Typ danych: Ciąg
  • Wymagany atrybut CloudEvent
  • Określa kontekst, w którym wystąpiło zdarzenie. Połączenie źródła i ID musi być unikalne dla każdego zdarzenia. Obecnie to pole jest zawsze wysyłane jako \/ zdarzenia streamowane z SQL.

• id

  • Typ danych: Ciąg
  • Wymagany atrybut CloudEvent
  • Identyfikuje zdarzenie. Producenci muszą zapewnić, że połączenie źródła i ID jest unikalne dla każdego konkretnego zdarzenia. Jeśli zduplikowane zdarzenie jest ponownie (na przykład z powodu błędu sieci) może mieć ten sam identyfikator. Konsumenci mogą założyć, że zdarzenia o identycznym źródle i identyfikatorze są duplikatami.

• logicalid

  • Typ danych: Ciąg
  • Atrybut rozszerzenia
  • Wspólne identyfikatory logiczne identyfikują podzielone wiadomości (ze względu na ograniczenia dotyczące rozmiaru wiadomości w Event Hubs).

• time

  • Typ danych: sygnatura czasowa
  • Opcjonalny atrybut CloudEvent
  • UTC znacznik czasu od momentu zatwierdzenia w transakcji SQL, która pierwotnie wywołuje zdarzenie streamowane.

• datacontenttype

  • Typ danych: Ciąg
  • Opcjonalny atrybut CloudEvent
  • Typ zawartości wartości danych. Ten atrybut umożliwia przenoszenie danych dowolnego typu zawartości, gdzie format i kodowanie mogą różnić się od formatu wybranego zdarzenia. Na przykład zdarzenie renderowane przy użyciu formatu koperty JSON może przenosić ładunek XML w danych, a odbiorca jest informowany przez ten atrybut ustawiony na "application/xml". Zasady renderowania treści danych dla różnych datacontenttype wartości są określone w specyfikacjach formatu zdarzeń.

• operation

  • Typ danych: Ciąg
  • Atrybut rozszerzenia
  • Reprezentuje typ operacji SQL, która miała miejsce:
    • INS dla insertów
    • UPD na aktualizacje
    • DEL dla usunięcia

• segmentindex

  • Typ danych: Liczba całkowita
  • Atrybut rozszerzenia
  • Indeks segmentów, który oznacza pozycję wiadomości w logicznym blokach wiadomości. Indeks segmentu zawiera informacje o tym, gdzie komunikat znajduje się w sekwencji fragmentów komunikatów logicznych. To pole jest zawsze obecne. Użyj logicalidpol , segmentindex, i finalsegment do sortowania nadchodzących zdarzeń reprezentujących duży ładunek SQL podzielony według skonfigurowanej max_message_size_kb wartości.

• finalsegment

  • Typ danych: wartość logiczna
  • Atrybut rozszerzenia
  • Wskazuje, czy ten segment jest ostatnim segmentem sekwencji. To pole jest zawsze obecne i pomaga określić, czy zdarzenie SQL zostało podzielone na podzdarzenia zgodnie z max_message_size_kb skonfigurowaną wartością.

• data

  • Typ danych: Tablica bajtów
  • Opcjonalny atrybut CloudEvent
  • Zawiera dane zdarzenia specyficzne dla danej dziedziny opisujące zmianę. Deserializuj bajty jako JSON lub Avro binary, zgodnie z wybranym formatem serializacji. Zdeserializowane dane podążają za schematem atrybutu danych CES Avro. Aby uzyskać informacje o jego polach, zobacz format atrybutów danych.

Przykłady

Przykład komunikatu JSON — wstawianie

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

Przykład wiadomości JSON - aktualizacja

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

Przykład komunikatu JSON — usuwanie

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

Format atrybutu danych

Atrybutem data jest tablica bajtów. Deserializuj bajty jako JSON lub Avro binary, zgodnie z wybranym formatem serializacji. W obu formatach wynikowy Data rekord podąża za schematem atrybutu danych CES Avro i zawiera dwa atrybuty:

• eventsource
• eventrow

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

Poniższe sekcje szczegółowo wyjaśniają zdeserializowane atrybuty.

źródło zdarzeń

Opisuje metadane dotyczące bazy danych i tabeli, w której wystąpiło zdarzenie:

• db

  • Typ danych: Ciąg
  • Opis: nazwa bazy danych, w której znajduje się tabela.
  • Przykład: EmployeesDb

• schema

  • Typ danych: Ciąg
  • Opis: schemat bazy danych zawierający tabelę.
  • Przykład: dbo

• tbl

  • Typ danych: Ciąg
  • Opis: tabela, w której wystąpiło zdarzenie.
  • Przykład: Employees

• cols

  • Typ danych: Tablica
  • Opis: Tablica zawierająca szczegóły kolumn w tabeli.
    • name (string): Nazwa kolumny.
    • type (string): Typ danych SQL kolumny, w tym jej długość, precyzja lub skala, jeśli to możliwe. Przykłady obejmują int, nvarchar(50)i datetime2(7).
    • index (liczba całkowita): Indeks lub pozycja kolumny w tabeli.

• pkkey

  • Typ danych: Tablica
  • Opis: reprezentuje kolumny klucza podstawowego i ich wartości identyfikujące konkretny wiersz.
    • columnname (string): Nazwa kolumny używanej w kluczu głównym.
    • value (string): Wartość kolumny używanej w kluczu głównym. Ta wartość pomaga jednoznacznie zidentyfikować wiersz.

• transaction

  • Typ danych: Obiekt
  • Opis: Opisuje transakcję SQL zawierającą operację danych.
    • commitlsn (string): Numer sekwencyjny zatwierdzeń (LSN) transakcji.
    • beginlsn (string): Początkowy LSN transakcji.
    • sequencenumber (liczba całkowita): Sekwencyjna liczba operacji danych w ramach transakcji. Użyj tej wartości do sortowania zdarzeń w ramach transakcji.
    • finalevent (boolean): Nie używa. To ciało zawsze ma wartość .false
    • committime (string): Data i godzina dokonania transakcji w bazie danych.

eventrow

Opisuje zmiany na poziomie wiersza i porównuje stare i bieżące wartości pól w rekordzie.

• old (obiekt opakowany w ciąg): reprezentuje wartości w wierszu przed zdarzeniem.
  • Każda para klucz-wartość składa się z:
    • <column_name>: (ciąg): nazwa kolumny.
    • <column_value>: (ciąg/int/etc.): poprzednia wartość dla tej kolumny.
• current (obiekt opakowany w ciąg): reprezentuje zaktualizowane wartości w wierszu po zdarzeniu.
  • Podobnie jak w przypadku starego obiektu, z każdą parą klucz-wartość ustrukturyzowaną jako:
    • <column_name> (ciąg): nazwa kolumny.
    • <column_value> (ciąg/int/etc.): nowa lub bieżąca wartość dla tej kolumny.

Schemat CES CloudEvent Avro

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

Schemat atrybutu danych CES Avro

Użyj następującego schematu podczas deserializacji tablicy data bajtów w natywnym JSON i binarnym CloudEvents:

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

Treści powiązane

• Co to jest zmiana przesyłania strumieniowego zdarzeń (wersja zapoznawcza)?
• Configure change event streaming (preview) do Azure Event Hubs
• Często zadawane pytania dotyczące programu CES