Настройка потоковой передачи событий изменений

Применимо к: Предварительная версия SQL Server 2025 (17.x)

В этой статье описывается, как настроить функцию потоковой передачи событий изменений (CES), представленную в предварительной версии SQL Server 2025 (17.x).

Обзор

Чтобы настроить и использовать потоковую передачу событий изменений, выполните следующую последовательность действий.

1. Используйте существующее или создайте новое пространство имен Azure Event Hubs и экземпляр Event Hubs. Узел Event Hubs получает события.
2. Включите потоковую передачу событий изменений для пользовательской базы данных.
3. Создайте группу потоков событий. С помощью этой группы настройте назначение, учетные данные, ограничения размера сообщения и схему секционирования.
4. Добавьте одну или несколько таблиц в группу потоков событий.

Каждый шаг подробно описан в следующих разделах этой статьи.

Предпосылки

Чтобы настроить потоковую передачу событий изменений, вам потребуется следующее:

• Пространство имен Центров событий Azure
• Экземпляр Центра событий Azure
• Имя узла Центров событий Azure
• Политика с уровнем доступа Отправить
• Вход, принадлежащий роли db_owner, или имеющий разрешение CONTROL DATABASE для базы данных, в которой планируется включить CES.

Настройка Центров событий Azure

Чтобы узнать, как создать центры событий Azure, просмотрите статью "Создание концентратора событий" с помощью портала Azure.

Чтобы настроить потоковую передачу в Центры событий Azure с помощью протокола AMQP (по умолчанию, собственного протокола Центров событий Azure), создайте маркер SAS для пространства имен Центров событий Azure и имени экземпляра. Это можно сделать программным способом с помощью любого языка программирования или скриптов. В этом примере показано, как создать маркер SAS из новой или существующей политики с помощью скрипта PowerShell.

Установка необходимых модулей

Чтобы управлять ресурсами Центров событий Azure с помощью скриптов PowerShell, необходимо иметь следующие модули:

• Модуль Az PowerShell
• Модуль Az.EventHub PowerShell

Следующий скрипт устанавливает необходимые модули:

Install-Module -Name Az -AllowClobber -Scope CurrentUser -Repository PSGallery -Force
Install-Module -Name Az.EventHub -Scope CurrentUser -Force

Если у вас уже есть необходимые модули и хотите обновить их до последней версии, выполните следующий сценарий:

Update-Module -Name Az -Force
Update-Module -Name Az.EventHub -Force

Подключение к Azure

Вы можете использовать Azure Cloud Shell или войти в систему и задать контекст подписки.

Чтобы запустить Azure Cloud Shell, просмотрите вход в Azure.

Определение политики

Чтобы создать маркер SAS, вам потребуется политика. Вы можете сделать одно из двух:

• Создайте новую политику с определенными правами.

  или

• Используйте существующую политику с правильными правами.

Создание маркера SAS для новой или существующей политики

При создании новой политики убедитесь, что у нее есть право на отправку. Если вы используете действующую политику, убедитесь, что она имеет право на отправку.

Следующий скрипт создаст новую политику или получит существующую, а затем создает из этой политики полный токен SAS в формате заголовка авторизации HTTP.

Замените значения в угловых скобках (<value>) значениями для вашей среды.

function Generate-SasToken {
$subscriptionId = "<Azure-Subscription-ID>"
$resourceGroupName = "<Resource-group-name>"
$namespaceName = "<Azure-Event-Hub-Namespace-name>"
$eventHubName = "<Azure-Event-Hubs-instance-name>"
$policyName = "<Policy-name>"

# Modifying the rest of the script is not necessary.

# Login to Azure and set Azure Subscription.
Connect-AzAccount

# Get current context and check subscription
$currentContext = Get-AzContext
if ($currentContext.Subscription.Id -ne $subscriptionId) {
    Write-Host "Current subscription is $($currentContext.Subscription.Id), switching to $subscriptionId..."
    Set-AzContext -SubscriptionId $subscriptionId | Out-Null
} else {
    Write-Host "Already using subscription $subscriptionId."
}

# Try to get the authorization policy (it should have Send rights)
$rights = @("Send")
$policy = Get-AzEventHubAuthorizationRule -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -EventHubName $eventHubName -AuthorizationRuleName $policyName -ErrorAction SilentlyContinue

# If the policy does not exist, create it
if (-not $policy) {
    Write-Output "Policy '$policyName' does not exist. Creating it now..."

    # Create a new policy with the Manage, Send and Listen rights
    $policy = New-AzEventHubAuthorizationRule -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -EventHubName $eventHubName -AuthorizationRuleName $policyName -Rights $rights
    if (-not $policy) {
        throw "Error. Policy was not created."
    }
    Write-Output "Policy '$policyName' created successfully."
} else {
    Write-Output "Policy '$policyName' already exists."
}

if ("Send" -in $policy.Rights) {
    Write-Host "Authorization rule has required right: Send."
} else {
    throw "Authorization rule is missing Send right."
}

$keys = Get-AzEventHubKey -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -EventHubName $eventHubName -AuthorizationRuleName $policyName

if (-not $keys) {
    throw "Could not obtain Azure Event Hub Key. Script failed and will end now."
}
if (-not $keys.PrimaryKey) {
    throw "Could not obtain Primary Key. Script failed and will end now."
}

# Get the Primary Key of the Shared Access Policy
$primaryKey = ($keys.PrimaryKey) 
Write-Host $primaryKey

## Check that the primary key is not empty.

# Define a function to create a SAS token (similar to the C# code provided)
function Create-SasToken {
    param (
        [string]$resourceUri, [string]$keyName, [string]$key
    )

$sinceEpoch = [datetime]::UtcNow - [datetime]"1970-01-01"
    $expiry = [int]$sinceEpoch.TotalSeconds + (60 * 60 * 24 * 31 * 6)  # 6 months
    $stringToSign = [System.Web.HttpUtility]::UrlEncode($resourceUri) + "`n" + $expiry
    $hmac = New-Object System.Security.Cryptography.HMACSHA256
    $hmac.Key = [Text.Encoding]::UTF8.GetBytes($key)
    $signature = [Convert]::ToBase64String($hmac.ComputeHash([Text.Encoding]::UTF8.GetBytes($stringToSign)))
    $sasToken = "SharedAccessSignature sr=$([System.Web.HttpUtility]::UrlEncode($resourceUri))&sig=$([System.Web.HttpUtility]::UrlEncode($signature))&se=$expiry&skn=$keyName"
    return $sasToken
}

# Construct the resource URI for the SAS token
$resourceUri = "https://$namespaceName.servicebus.windows.net/$eventHubName"

# Generate the SAS token using the primary key from the new policy
$sasToken = Create-SasToken -resourceUri $resourceUri -keyName $policyName -key $primaryKey

# Output the SAS token
Write-Output @"
-- Generated SAS Token --
$sasToken
-- End of generated SAS Token --
"@
}

Generate-SasToken

Включение и настройка потоковой передачи событий изменений

Чтобы включить и настроить потоковую передачу событий изменений, измените контекст базы данных на пользовательную базу данных и выполните следующие действия.

1. Если она еще не настроена, задайте для базы данных полную модель восстановления.
2. Создайте мастер-ключ и учетные данные с ограничением на уровень базы данных.
3. Включите потоковую передачу событий.
4. Создайте группу потоков событий.
5. Добавьте одну или несколько таблиц в группу потоков событий.

В примерах этого раздела показано, как включить CES для протокола AMQP и протокола Apache Kafka.

Ниже приведены примеры значений параметров для примеров в этом разделе:

• @stream_group_name = N'myStreamGroup'
• @destination_location = N'myEventHubsNamespace.servicebus.windows.net/myEventHubsInstance'
• @partition_key_scheme = N'None'
• Значение первичного или вторичного ключа: Secret = 'BVFnT3baC/K6I8xNZzio4AeoFt6nHeK0i+ZErNGsxiw='
• EXEC sys.sp_add_object_to_event_stream_group N'myStreamGroup', N'dbo.myTable'

Пример: Передача данных в Центры событий Azure с использованием протокола AMQP (аутентификация с помощью маркера SAS)

Замените значения в угловых скобках (<value>) значениями для вашей среды.

USE <database name>

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

CREATE DATABASE SCOPED CREDENTIAL <CredentialName>
    WITH IDENTITY = 'SHARED ACCESS SIGNATURE',
    SECRET = '<Generated SAS Token>'

EXEC sys.sp_enable_event_stream

EXEC sys.sp_create_event_stream_group
    @stream_group_name =      N'<EventStreamGroupName>',
    @destination_type =       N'AzureEventHubsAmqp',
    @destination_location =   N'<AzureEventHubsHostName>/<EventHubsInstance>',
    @destination_credential = <CredentialName>,
    @max_message_size_kb =    <MaxMessageSize>, 
    @partition_key_scheme =   N'<PartitionKeyScheme>'

EXEC sys.sp_add_object_to_event_stream_group
    N'<EventStreamGroupName>',
    N'<SchemaName>.<TableName>'

Пример: Передача данных в Центры событий Azure через протокол AMQP (аутентификация по ключевому значению)

Замените значения в угловых скобках (<value>) значениями для вашей среды.

USE <database name>

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

CREATE DATABASE SCOPED CREDENTIAL <CredentialName>
    WITH IDENTITY = '<Azure Event Hubs SAS Policy name>',
    SECRET = '<Primary or Secondary key value>'

EXEC sys.sp_enable_event_stream

EXEC sys.sp_create_event_stream_group
    @stream_group_name =      N'<EventStreamGroupName>',
    @destination_type =       N'AzureEventHubsAmqp',
    @destination_location =   N'<AzureEventHubsHostName>/<EventHubsInstance>',
    @destination_credential = <CredentialName>,
    @max_message_size_kb =    <MaxMessageSize>,
    @partition_key_scheme =   N'<PatitionKeyScheme>'

EXEC sys.sp_add_object_to_event_stream_group
    N'<EventStreamGroupName>',
    N'<SchemaName>.<TableName>'

Пример. Потоковая передача в Центры событий Azure с помощью протокола Apache Kafka (проверка подлинности строки подключения)

Замените значения в угловых скобках (<value>) значениями для вашей среды.

USE <database name>

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

CREATE DATABASE SCOPED CREDENTIAL credential1
    WITH IDENTITY = 'SHARED ACCESS SIGNATURE',
    SECRET = '<Event Hubs Namespace – Primary or Secondary connection string>'

EXEC sys.sp_enable_event_stream

EXEC sys.sp_create_event_stream_group
    @stream_group_name =      N'<EventStreamGroupName>',
    @destination_type =       N'AzureEventHubsApacheKafka',
    @destination_location =   N'<AzureEventHubsHostName>:<port>/<EventHubsInstance>',
    @destination_credential = <CredentialName>,
    @max_message_size_kb =    <MaxMessageSize>,
    @partition_key_scheme =   N'<PatitionKeyScheme>'

EXEC sys.sp_add_object_to_event_stream_group
    N'<EventStreamGroupName>',
    N'<SchemaName>.<TableName>'

Пример. Потоковая передача в Центры событий Azure с помощью протокола Apache Kafka (проверка подлинности значения ключа)

Замените значения в угловых скобках (<value>) значениями для вашей среды.

USE <database name>

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

CREATE DATABASE SCOPED CREDENTIAL credential1
    WITH IDENTITY = '<Azure Event Hubs SAS Policy name>',
    SECRET = '<Primary or Secondary key value>' -- BVFnT3baC/K6I8xNZzio4AeoFt6nHeK0i+ZErNGsxiw=

EXEC sys.sp_enable_event_stream

EXEC sys.sp_create_event_stream_group
    @stream_group_name =      N'<EventStreamGroupName>',  -- myStreamGroup
    @destination_type =       N'AzureEventHubsApacheKafka',
    @destination_location =   N'<AzureEventHubsHostName>:<port>/<EventHubsInstance>', -- myEventHubsNamespace.servicebus.windows.net:9093/myEventHubsInstance
    @destination_credential = <CredentialName>,
    @max_message_size_kb =    <MaxMessageSize>,       -- 1024
    @partition_key_scheme =   N'<PatitionKeyScheme>'  -- N'None'

EXEC sys.sp_add_object_to_event_stream_group
    N'<EventStreamGroupName>',
    N'<SchemaName>.<TableName>' -- dbo.myTable

Просмотр конфигурации и функции CES

В sys.databasesis_event_stream_enabled = 1 указывает, что потоковая передача событий изменений включена для базы данных.

Следующий запрос возвращает все базы данных с включенной потоковой передачей событий изменений:

SELECT * FROM sys.databases WHERE is_event_stream_enabled = 1

В sys.tablesis_replicated = 1 означает, что таблица передается в поток, а sp_help_change_feed_table предоставляет информацию о группе таблиц и метаданных таблицы для потоковой передачи событий изменений.

Следующий запрос возвращает все таблицы с включенной потоковой передачей событий изменений и предоставляет сведения о метаданных:

SELECT name, is_replicated FROM sys.tables

EXEC sp_help_change_feed_table @source_schema = '<schema name>', @source_name = '<table name>'

Хранимые процедуры CES, системные функции и динамические административные представления

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

Ограничения

Потоковая передача событий изменений (CES) имеет следующие ограничения:

Ограничения уровня сервера и общие ограничения

• CES не поддерживается в SQL Server 2025 в Linux или выпуске SQL Server 2025 Express.
• CES выдает события только для изменений данных из INSERT, UPDATE и DELETE инструкций DML.
• CES не обрабатывает изменения схемы (операции DDL), что означает, что он не будет выдавать события для операций DDL. Однако операции DDL не блокируются, поэтому при выполнении схема последующих событий DML отражает обновленную структуру таблицы. Ожидается, что пользователи будут корректно обрабатывать события с обновленной схемой.
• Если JSON является указанным форматом выходных данных, сообщения больших событий могут быть разделены примерно на 25% настроенного максимального размера сообщения для каждой группы потоков. Это ограничение не применяется к двоичному типу выходных данных.
• Если сообщение превышает ограничение размера сообщения Центров событий Azure, то сбой в настоящее время наблюдается только с помощью расширенных событий.
• Переименования таблиц и столбцов блокируются для таблиц, настроенных для CES. Разрешены переименования базы данных.

Ограничения уровня базы данных

• CES поддерживается только в базах данных, настроенных с помощью полной модели восстановления.
• CES не поддерживается в базах данных, настроенных с помощью зеркальных баз данных Fabric для SQL Server, репликации транзакций, отслеживания измененных данных или Azure Synapse Link. Отслеживание изменений поддерживается в базах данных, настроенных с помощью CES.
• CES может транслировать только из записываемых основных баз данных. Вторичные базы данных, которые являются частью групп доступности Always On или которые используют ссылку управляемого экземпляра, не могут быть настроены в качестве источников потоковой передачи.
• CES нельзя включить для представлений или материализованных представлений.

Ограничения на уровне таблицы

• Таблица может принадлежать только одной группе потоковой передачи. Потоковая передача одной таблицы в несколько назначений не поддерживается.
• Для CES можно настроить только пользовательские таблицы. Системные таблицы не поддерживаются.
• Можно настроить до 4 096 групп потоков. Каждая группа потоков может включать до 40 000 таблиц.
• Хотя в таблице активирована CES, ограничение на первичный ключ нельзя добавить в таблицу или удалить из нее.
• ALTER TABLE SWITCH PARTITION не поддерживается в таблицах, настроенных для CES.
• TRUNCATE TABLE не поддерживается в таблицах, активированных для CES.
• CES не поддерживает таблицы, использующие какие-либо из следующих функций:
  • Кластеризованные индексы Columnstore
  • Временные таблицы журнала или таблицы журнала реестра
  • Всегда зашифровано
  • OLTP в памяти (оптимизированные для памяти таблицы)
  • Таблицы графов
  • Внешние таблицы

Ограничения на уровне столбцов

• Следующие типы данных не поддерживаются CES. Столбцы этих типов пропускаются при стриминге:
  • json
  • image
  • text / ntext
  • xml
  • rowversion / timestamp
  • sql_variant
  • Определяемые пользователем типы (UDT)
  • geometry
  • geography
  • vector

Разрешения в исходной базе данных

• Для обеспечения безопасности на уровне строк CES выдает изменения со всех строк независимо от разрешений пользователя.
• Динамическое маскирование данных не применяется к данным, отправленным через CES. Данные передаются в поток без маскировки, даже если настроена маскировка.
• CES не выдает события, связанные с изменениями разрешений на уровне объекта (например, предоставление разрешений определенным столбцам).

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

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