將 SQL 變更事件串流至 Eventstream，以便 Real-Time 處理

變更事件串流（CES）功能是 SQL Server 2025（17.x）及 Azure SQL Database 引入的一項新進階資料整合功能。它允許你直接從 SQL Server 傳送即時資料變更事件到 Azure 事件中樞。 Fabric 事件串流功能與 Azure 事件中樞相關聯，後者是一個完全託管的雲端服務。當你建立事件串流時，事件集線中心命名空間會自動被配置，事件集線器也會被分配給預設串流。這表示 CES 也能將變更事件直接傳送到 Fabric Eventstream。

本教學示範如何利用變更事件串流（CES）功能，將 Azure SQL Database 的 SQL 變更事件串流到 Fabric Eventstream 中的自訂端點。然後透過 Eventstream 的 SQL 運算子，從變更事件中擷取 SQL 資料表資料。

在本教學課程中，您將瞭解如何：

建立事件串流並新增自訂端點來源
從自訂端點來源取得連線資訊
配置 Azure SQL 資料庫中的變更事件串流
使用 Eventstream SQL 運算子從變更事件中擷取資料表資料

先決條件

取得一個有貢獻者或更高權限的工作區，該工作區位於你的事件串流所在地。若使用 Microsoft Entra 認證，則需擁有會員或以上權限才能授予 Entra ID 貢獻者或更高權限。
若要啟用 CES，必須以擁有 db_owner 角色或對 Azure SQL 資料庫具有 CONTROL DATABASE 權限的帳戶登入。
對於已配置使用出站防火牆規則或網路安全邊界的 Azure SQL Database，應允許透過開放的防火牆埠口存取 Eventstream 的自訂來源端點。

建立事件流

前往 Fabric 入口網站。
選取左側導覽列上的 [我的工作區 ]。
在 我的工作區 頁面上，選取命令列上的 + 新增專案 。
在 [新增專案] 頁面上，搜尋 Eventstream，然後選取 [Eventstream]。
在 [新增事件串流] 視窗中，輸入事件串流的名稱，然後選取 [建立]。
在您的工作區中建立新的事件串流可能需要幾秒鐘的時間。建立 Eventstream 之後，會將您導向主要編輯器，在其中您可以開始將來源新增至 Eventstream。

從新增的自訂端點來源取得連線資訊

要從事件串流取得連線端點資訊，請在事件流中新增一個自訂端點來源。連線端點隨後會隨時可用並暴露於自訂端點來源中

要新增自訂端點來源，請在開始頁面選擇「使用自訂端點」
在 自訂端點 對話框中，輸入來源名稱下的自訂來源 名稱，然後選擇 新增。
在您建立自訂端點來源之後，它將被新增至編輯模式下的畫布上的事件流中。要實作自訂應用程式來源新增的資料，請選擇發佈
事件串流發佈後，請在畫布上選擇自訂端點來源圖塊以查看其詳細資訊。底部窗格提供配置 Azure SQL 資料庫中 CES 功能的端點資訊。

啟用並配置 Azure SQL Database 中的變更事件串流

本節將引導您如何啟用並設定 Azure SQL 資料庫的變更事件串流。首先選擇並登入你想啟用此功能的特定資料庫，然後依照以下步驟操作：

如果資料庫還沒設定，請設成完整的復原模型。
建立主要金鑰和資料庫範圍認證。
啟用事件串流。
建立事件數據流群組。
將一或多個數據表新增至事件數據流群組。

變更事件的串流支援兩種傳輸協定，將變更事件輸送到事件流中：

AMQP 協定
Kafka 通訊協定

兩種方法都需要完成這些步驟，但參數不同。而且兩種方式都支援三種認證類型：Microsoft Entra、SAS 憑證，以及 SAS 金鑰。你可以選擇最適合你需求的協定和認證類型。在以下範例中，我們使用 AMQP 協議 以及 Microsoft Entra 以及 SAS 令牌認證來逐步完成這些步驟。關於其他協定及認證方法，請參考「配置變更事件串流」

1. 如果資料庫尚未設定為完整復原模型，請將其設定為完整復原模式：

此步驟僅在使用 SQL Server 2025 時必要。對於 Azure SQL Database 或 Azure SQL 受控執行個體，設定永遠是 FULL，因此此步驟可跳過。

-- set the recovery mode as FULL
USE <database name>;
GO
ALTER DATABASE <database name>
SET RECOVERY FULL;
GO
-- Check the recovery mode
SELECT name, recovery_model_desc
FROM sys.databases
WHERE name = '<database name>';
GO

2. 建立主金鑰與資料庫範圍驗證憑證：

-- Create the Master Key with a password.
CREATE MASTER KEY ENCRYPTION BY PASSWORD = '<Password>';

3. 啟用事件串流：

-- Enable change event streaming for the selected database.
EXEC sys.sp_enable_event_stream;

4. 建立活動串流群組：

要建立事件串流群組，你需要決定使用哪一種認證。推薦兩種認證類型：Microsoft Entra 和 SAS 令牌。要使用 AMQP 協定在 SQL 資料庫建立變更事件串流群組，請使用 Eventstream 原始自訂端點的金鑰資訊進行設定與設定。

<EventHubNamespaceEndpoint> 與 <EventHubName>
<SharedAccessKeyName> 與 <PrimaryOrSecondaryKey>

Microsoft Entra

這是最安全的認證方式，強烈建議。請依照以下步驟來設定 CES：

在 Azure 入口網站啟用您的 Azure SQL 資料庫的系統分配的受控身分：Security>Identity>System assigned managed identity：
在你的Fabric工作區中，授予你的Azure SQL資料庫系統分配的管理身份 Contributor 或更高的權限，該工作區的事件流所在區域包括：Workspace>Manage access>Add people or groups，請輸入你的 Azure SQL 實例名稱以進行搜尋：

登入您的 Azure SQL 資料庫並執行以下 SQL 指令。將所有 <> 值替換為你事件串流和 SQL 環境特有的值。

CREATE DATABASE SCOPED CREDENTIAL <CredentialName>
WITH IDENTITY = 'Managed Identity'

EXEC sys.sp_create_event_stream_group
    @stream_group_name = N'<ChangeEventStreamGroupName>',
    @destination_type = N'AzureEventHubsAmqp',
    @destination_location = N'<EventHubNamespaceEndpoint>/<EventHubName>', -- Event hub namespace endpoint should NOT contain the port number: 9093
    @destination_credential = <CredentialName>,
    @max_message_size_kb = <MaxMessageSize>, -- 128, 256, 1024
    @partition_key_scheme = N'<PartitionKeyScheme>', -- Possible options: N'None' (partition is chosen in “round robin”, N'StreamGroup', N'Table' (partitioning by     table), N'Column'. If nothing is supplied, the default is None
    @encoding = N'JSON'

SAS 權杖

如果 Microsoft Entra 無法驗證，請使用 SAS 令牌認證。請依照以下步驟設定 CES 並使用此認證：

用腳本產生 SAS 代幣，並輸入你的事件串流端點資訊。以下是使用 PowerShell 的範例。如果你想在其他語言中使用腳本，請參考「產生共享存取簽章令牌」。將中的 <> 值替換成你事件串流端點提供的值。

[Reflection.Assembly]::LoadWithPartialName("System.Web") | out-null
$URI="<EventHubNamespaceEndpoint>/<EventHubName>/"
$Access_Policy_Name="<SharedAccessKeyName>"
$Access_Policy_Key="<PrimaryOrSecondaryKey>"
#Token expires now+ 86400(1 day)
$Expires=([DateTimeOffset]::Now.ToUnixTimeSeconds())+86400
$SignatureString=[System.Web.HttpUtility]::UrlEncode($URI)+ "`n" +   [string]  $Expires
$HMAC = New-Object System.Security.Cryptography.HMACSHA256
$HMAC.key = [Text.Encoding]::ASCII.GetBytes($Access_Policy_Key)
$Signature = $HMAC.ComputeHash([Text.Encoding]::ASCII.GetBytes    ($SignatureString))
$Signature = [Convert]::ToBase64String($Signature)
$SASToken = "SharedAccessSignature sr=" + [System.Web.HttpUtility]    ::UrlEncode($URI) + "&sig=" + [System.Web.HttpUtility]::UrlEncode    ($Signature) + "&se=" + $Expires + "&skn=" + $Access_Policy_Name
$SASToken

登入您的 Azure SQL 資料庫並執行以下 SQL 指令。請將所有 <> 值替換成你事件串流和 SQL 環境特有的值

CREATE DATABASE SCOPED CREDENTIAL <CredentialName>
WITH IDENTITY = 'SHARED ACCESS SIGNATURE',
SECRET = '<Generated SAS Token>' -- Be sure to copy the entire token   from   previous step. The SAS token starts with "SharedAccessSignature   sr="

EXEC sys.sp_create_event_stream_group
    @stream_group_name = N'<ChangeEventStreamGroupName>',
    @destination_type = N'AzureEventHubsAmqp',
    @destination_location = N'<EventHubNamespaceEndpoint>/  <EventHubName>',   -- Event hub namespace endpoint should NOT   contain the port number:   9093
    @destination_credential = <CredentialName>,
    @max_message_size_kb = <MaxMessageSize>, -- 128, 256, 1024
    @partition_key_scheme = N'<PartitionKeyScheme>', -- Possible   options:   N'None' (partition is chosen in “round robin”,   N'StreamGroup',   N'Table' (partitioning by table), N'Column'. If   nothing is supplied,   the default is None
    @encoding = N'JSON'

5. 將一個或多個資料表加入事件串流群組

-- add table into the stream group created above
EXEC sys.sp_add_object_to_event_stream_group
@stream_group_name = N'<ChangeEventStreamGroupName>',
@object_name = N'<SchemaName>.<TableName>'

在依上述步驟設定好 Azure SQL 資料庫中的變更事件串流後，你就可以開始修改資料庫了。要確認這些變更是否流入 Eventstream，請前往 Eventstream 並選擇預設串流節點（中間節點）。例如，新增紀錄加入表格，或刪除現有紀錄，如下所示：

使用 SQL 運算子從變更事件中擷取資料表資料

要從變更事件中提取 SQL 資料表資料，了解事件格式及每欄的資料型態非常重要。 SQL 變更事件遵循 CloudEvents 規範。實際的資料表資料包含在「data」欄位，而「操作」欄位則指定該紀錄是否已入、更新或刪除。欲了解更多資訊，請參閱 JSON 訊息格式。

以下範例示範如何透過在事件串流中處理變更事件，從 SQL 表格「採購」中提取採購訂單資料。這些擷取出來的資料可用來分析新的訂單資訊。

EXEC sp_help 'dbo.Purchases';

根據 JSON 訊息格式中描述的變更事件格式，change event 中的資料表資料位於「eventrow」節點的「data」欄位，其型別為「string」，遵循 JSON 格式，也就是 JSON 字串，可解析為 JSON。 Eventstream 中的 SQL 運算子內建函式：json_parse（string），可以將 JSON 字串解析成 JSON 物件。請參考下方資料欄範例的截圖。
由於只需新訂單的資料，因此只需包含帶有「插入」操作的變更事件，即帶有「INS」的「操作」欄位：「操作」=「INS」。

在事件串流的編輯模式中，新增一個「SQL 運算子」來擷取資料，並加入一個衍生串流來接收擷取後的資料。

以 SQL 運算子開啟 SQL 編輯器，並輸入以下 SQL 語言：

WITH PurchaseOrders AS
(
    SELECT json_parse(data).eventrow.[current] AS neworders
    FROM [eventstream-sqlces-stream]
    WHERE operation = 'INS'
)
SELECT
    json_parse(neworders).ID AS ID,
    json_parse(neworders).Name AS Name,
    json_parse(neworders).PurchaseDate AS PurchaseDate,
    json_parse(neworders).Product AS Product,
    json_parse(neworders).ProductCategory AS ProductCategory,
    json_parse(neworders).OrderValue AS OrderValue,
    json_parse(neworders).LoyaltyProgramMember AS LoyaltyProgramMember,
    json_parse(neworders).NPSScore AS NPSScore
INTO [DerivedStream]
FROM [PurchaseOrders]

使用 query test 預覽此 SQL 查詢的輸出，如下所示。

完成設定並發布事件串流後，您可以在事件串流的即時檢視中選擇衍生串流，查看最新的訂單資料。

這個教學展示了如何透過 Eventstream 的來源自訂端點，利用 Azure SQL 資料庫中新引入的變更事件串流功能，將 SQL 變更事件串流到你的事件串流。它也會教你如何用 SQL 運算子從變更事件中擷取資料表資料。資料進入事件串流並擷取後，你可以將其導向不同目的地進行分析、警示與報告。以下是一些有助於進一步參考的資源：

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-04-10