JSON 消息格式 - 更改事件流式处理

适用于： SQL Server 2025 （17.x）Azure SQL 数据库 Azure SQL 托管实例 Microsoft Fabric 中的 SQL 数据库

本文介绍了当你使用SQL Server 2025（17.x）、Azure SQL 数据库、Azure SQL 托管实例中change event streaming（CES）功能时，流向Azure 事件中心或Fabric Eventstream的CloudEvents消息格式以及Microsoft Fabric中的SQL数据库。

注释

变更事件流目前处于预览阶段，不同产品在支持性上存在差异。在预览期间，此功能可能会更改。

概述

变更事件流会发出符合 CloudEvents 规范的事件，因此你可以轻松将它们集成到事件驱动系统中。所有 CES CloudEvents 都包含 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}表示Microsoft变更事件流式DML事件模式的版本。
  - 当前最新的模式版本是1.
source
- 数据类型：字符串
- 必需的 CloudEvent 属性
- 标识发生事件的上下文。源和ID的组合必须为每个事件唯一。目前，该字段总是像事件一样从SQL流式发送 \/ 。
id
- 数据类型：字符串
- 必需的 CloudEvent 属性
- 标识事件。制作方必须确保源与ID的组合在每个不同事件中都是唯一的。如果重复事件重新发送（例如，由于网络错误），它可能具有相同的 ID。使用者可能假定具有相同源和 ID 的事件是重复的。
logicalid
- 数据类型：字符串
- 扩展属性
- 共享逻辑ID用于识别分裂消息（由于事件中心的消息大小限制）。
time
- 数据类型：时间戳
- 可选 CloudEvent 属性
- 提交发生在最初触发流事件的SQL事务中的时间戳。
datacontenttype
- 数据类型：字符串
- 可选 CloudEvent 属性
- 数据类型值。此属性使数据能够携带任何类型的内容，其中格式和编码可能与所选事件格式的不同。例如，使用 JSON 信封格式呈现的事件可能会在数据中携带 XML 有效负载，使用者会通过此属性设置为“application/xml”来通知使用者。不同值的数据内容渲染 datacontenttype 规则在事件格式规范中定义。
operation
- 数据类型：字符串
- 扩展属性
- 表示发生的SQL作类型：
  - 插页的INS
  - 更新
  - 删除时的 DEL
segmentindex
- 数据类型：整数
- 扩展属性
- 段索引，表示消息在逻辑消息块中的位置。段索引提供有关消息在逻辑消息片段序列中的位置的信息。这个场始终存在。使用 logicalid、 segmentindex和 finalsegment 字段来排序代表大SQL负载分配的输入事件，这些事件根据配置 max_message_size_kb 值进行排序。
finalsegment
- 数据类型：布尔值
- 扩展属性
- 表示该段是否为序列的最后一段。该字段始终存在，有助于识别SQL事件是否根据配置 max_message_size_kb 值被拆分为子事件。
data
- 数据类型：字节数组
- 可选 CloudEvent 属性
- 包含描述变更的领域特定事件数据。根据选定的序列化格式，将字节反序列化为JSON或Avro二进制。反序列化数据遵循 CES数据属性Avro模式。有关其字段的信息，请参见数据属性格式。

注释

消息拆分与列值截断是分开的。在 CES 序列化该 data 属性之前，会将每个大于 1 MB 的流列值截断为 1 MB。 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
- 数据类型：字符串
- 说明：表所在的数据库的名称。
- 示例：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 二进制 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