適用於：
SQL Server 2025 （17.x） Azure SQL DatabaseAzure SQL 受控執行個體Microsoft Fabric 中的 SQL 資料庫
本文說明了當你在 SQL Server 2025（17.x）、Azure SQL Database、Azure SQL 受控執行個體 中使用變更事件串流（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 屬性
- 包含值，描述與原始發生事件相關的事件類型。 此值的格式由生產者定義，可能包含類型版本等資訊。 更多資訊請參閱 CloudEvent 版本管理。
- 對於變更事件串流事件，目前的類型為：
com.microsoft.SQL.CES.DML.V{n}，其中{n}表示 Microsoft 變更事件串流 DML 事件結構的版本。- 目前最新的結構版本為 1.
source- 數據類型：字串
- 必要 CloudEvent 屬性
- 識別發生事件的內容。 來源與ID的組合必須為每個事件唯一。 目前，這個欄位總是像從 SQL 串流的事件一樣傳送
\/。
id- 數據類型：字串
- 必要 CloudEvent 屬性
- 識別事件。 製作人必須確保來源與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。 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 訊息範例 - delete
{
"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 架構 ，並包含兩個屬性：
eventsourceeventrow
{
"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>（string）：數據行的名稱。 -
<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"
}
]
}
}
]
}