Формат сообщения JSON — изменение потоковой передачи событий

Применимо к: SQL Server 2025 (17.x) База данных Azure SQLУправляемый экземпляр Azure SQLБаза данных SQL в Microsoft Fabric

В этой статье описывается формат сообщений CloudEvents, который передаётся в Центры событий Azure или Fabric Eventstream при использовании функции потока событий изменения (CES) в SQL Server 2025 (17.x), База данных SQL Azure, Управляемый экземпляр SQL Azure, и SQL базы данных в Microsoft Fabric.

Замечание

Потоковая трансляция событий изменений сейчас находится в предварительном просмотре и отличается в поддержке между продуктами. Во время предварительной версии эта функция подлежит изменению.

Обзор

Поток событий изменений транслирует события, соответствующие спецификации CloudEvents , поэтому их можно легко интегрировать с системами, управляемыми событиями. Все ceS CloudEvents содержат 11 атрибутов (полей). Вы можете настроить CES так, чтобы сериализация всего CloudEvent, включая атрибут data , как нативный JSON или Avro бинарный файл. Нативные JSON-события не содержат бинарных секций Avro. В обоих форматах data сериализации атрибут имеет тип байт-массива. Байты используют бинарное кодирование JSON или Avro согласно выбранному формату сериализации и следуют схеме атрибута данных CES Avro.

Important

С 15 августа 2026 года протокол AMQP устарел для потокового потокового сообщения событий изменений (CES). Существуют различия между платформами. Для шагов и сроков миграции см. устарение протокола AMQP.

Когда это применимо, описания в этом разделе взяты из спецификации CloudEvent, содержащей более подробную информацию.

Атрибуты

  • specversion:

    • Тип данных: String
    • Обязательный атрибут CloudEvent
    • Версия спецификации CloudEvents, используемой событием. Эта версия позволяет интерпретировать контекст.

  • type

    • Тип данных: String
    • Обязательный атрибут CloudEvent
    • Содержит значение, описывающее тип события, связанного с исходной вхождением. Формат этого значения определяется продюсером и может включать информацию, например, версию типа. Для получения дополнительной информации см. раздел Versioning of CloudEvents.
    • Для потоковых событий изменения событий в настоящее com.microsoft.SQL.CES.DML.V{n}время тип: , где {n} обозначает версию схемы DML-события Microsoft для потокового события изменений.
      • Текущая последняя версия схемы — 1.

  • source

    • Тип данных: String
    • Обязательный атрибут CloudEvent
    • Определяет контекст, в котором произошло событие. Комбинация источника и ID должна быть уникальной для каждого события. В настоящее время это поле всегда отправляется как \/ события, транслируемые из SQL.

  • id

    • Тип данных: String
    • Обязательный атрибут CloudEvent
    • Определяет событие. Продюсеры должны гарантировать, что комбинация источника и ID уникальна для каждого отдельного события. Если повторяющееся событие повторно (например, из-за сетевой ошибки) может иметь тот же идентификатор. Потребители могут предположить, что события с идентичным источником и идентификатором дублируются.

  • logicalid

    • Тип данных: String
    • Атрибут расширения
    • Общие логические идентификаторы идентифицируют разделённые сообщения (из-за ограничений по размеру сообщений в Event Hubs).

  • time

    • Тип данных: метка времени
    • Необязательный атрибут CloudEvent
    • UTC временная метка, когда коммит произошёл в рамках SQL-транзакции, которая изначально запускает потоковое событие.

  • datacontenttype

    • Тип данных: String
    • Необязательный атрибут CloudEvent
    • Тип контента значения данных. Этот атрибут позволяет данным переносить любой тип содержимого, в котором формат и кодировка могут отличаться от формата выбранного события. Например, событие, отображаемое с помощью формата конверта JSON, может содержать полезные данные XML в данных, и потребитель сообщает этому атрибуту значение application/xml. Правила отображения содержимого данных для разных datacontenttype значений определены в спецификациях формата событий.

  • operation

    • Тип данных: String
    • Атрибут расширения
    • Представляет тип SQL-операции, которая произошла:
      • INS для вставок
      • UPD для обновлений
      • DEL для удалений

  • segmentindex

    • Тип данных: целое число
    • Атрибут расширения
    • Индекс сегментов, который обозначает положение сообщения внутри логических блоков сообщений. Индекс сегмента содержит сведения о том, где сообщение находится в последовательности фрагментов логического сообщения. Это поле всегда присутствует. Используйте logicalid, , и segmentindex поля для сортировки входящих событий, представляющих собой большое разделение SQL-нагрузки, по заданному finalsegmentmax_message_size_kbзначению.

  • finalsegment

    • Тип данных: Логический
    • Атрибут расширения
    • Указывает, является ли этот сегмент последним сегментом. Это поле всегда присутствует и помогает определить, было ли SQL-событие разделено на подсобытия в соответствии с установленным max_message_size_kb значением.

  • data

    • Тип данных: байтовый массив
    • Необязательный атрибут CloudEvent
    • Содержит данные, специфичные для домена, описывающие изменения. Десериализуйте байты как JSON или Avro бинар в соответствии с выбранным форматом сериализации. Десериализированные данные следуют схеме атрибута данных CES Avro. Для информации о его полях см. формат атрибута данных.

Замечание

Разделение сообщений происходит отдельно от усечения столбцов. Перед сериализацией атрибута CES data урезает значение каждого потокового столбца больше 1 МБ до 1 МБ. Затем 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 и содержит два атрибута:

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

В следующих разделах более подробно объясняются десериализированные атрибуты.

eventsource

Описывает метаданные базы данных и таблицы, в которой произошло событие:

  • db

    • Тип данных: String
    • Описание: имя базы данных, в которой находится таблица.
    • Пример: EmployeesDb

  • schema

    • Тип данных: String
    • Описание: схема базы данных, содержащая таблицу.
    • Пример: dbo

  • tbl

    • Тип данных: String
    • Описание: таблица, в которой произошло событие.
    • Пример: Employees

  • cols

    • Тип данных: Массив
    • Описание: массив, подробный сведения о столбцах в таблице.
      • name (string): Название колонки.
      • type (строка): Тип данных SQL, включая его длину, точность или масштаб, когда это применимо. Например, int, nvarchar(50) и datetime2(7).
      • index (целое число): Индекс или положение столбца в таблице.

  • pkkey

    • Тип данных: Массив
    • Описание. Представляет столбцы первичного ключа и их значения для идентификации конкретной строки.
      • columnname (строка): Название столбца, используемого в первичном ключе.
      • value (строка): Значение столбца, используемого в первичном ключе. Это значение помогает уникально идентифицировать строку.

  • transaction

    • Тип данных: Объект
    • Описание: Описывает SQL-транзакцию, содержащую операцию с данными.
      • commitlsn (строка): Коммит-лог последовательного номера (LSN) транзакции.
      • beginlsn (строка): Начальная LSN транзакции.
      • sequencenumber (целое число): Последовательное число операции с данными внутри транзакции. Используйте это значение для сортировки событий внутри транзакции.
      • finalevent (булевый): Не используется. Это поле всегда имеет значение .false
      • committime (строка): Дата и время, когда транзакция была зафиксирована в базе данных.

Замечание

В продуктах SQL, настроенных с часовым поясом, не связанным с UTC, 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 Avro schema

Используйте следующую схему при десериализации data массива байтов в нативных JSON и Avro бинарных 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"
          }
        ]
      }
    }
  ]
}

Связанный контент

  • Last updated on