Transmisión de eventos de cambio de SQL a Eventstream para el procesamiento en tiempo real

La característica Change Event Streaming (CES) es una nueva funcionalidad de integración de datos avanzada introducida en SQL Server 2025 (17.x) y Azure SQL Database. Permite enviar eventos de cambio de datos en tiempo real de SQL Server directamente a Azure Event Hubs. La característica eventstreams de Fabric está asociada a Azure Event Hubs, que es un servicio totalmente administrado basado en la nube. Al crear una secuencia de eventos, se aprovisiona automáticamente un espacio de nombres del centro de eventos y se asigna un centro de eventos a la secuencia predeterminada. Esto significa que CES también puede entregar eventos de cambio directamente a Fabric Eventstream.

En este tutorial se muestra cómo transmitir los eventos de cambio de SQL de Azure SQL Database a un punto de conexión personalizado en Fabric Eventstream mediante la característica Change Event Streaming (CES). A continuación, extraiga los datos de la tabla SQL de los eventos de cambio a través del operador SQL de Eventstream.

En este tutorial aprenderá a:

  • Creación de una secuencia de eventos y adición de un origen de punto de conexión personalizado
  • Obtén la información de conexión de una fuente de endpoint personalizada.
  • Configuración de la transmisión de eventos de cambios en la base de datos de Azure SQL
  • Extracción de datos de tabla de eventos de cambio mediante el operador SQL Eventstream

Prerrequisitos

  • Acceda a un espacio de trabajo con permisos de Colaborador o superiores donde se encuentra el flujo de eventos. Si se usa la autenticación de Microsoft Entra, se requieren permisos de miembro o superiores para conceder acceso de colaborador de Entra ID o superior.
  • Inicio de sesión en el rol db_owner o que tenga permiso de CONTROL DATABASE para la base de datos de Azure SQL donde pretende habilitar CES.
  • Para Azure SQL Database configurado para usar reglas de firewall de salida o un Perímetro de Seguridad de Red, permita el acceso al punto de conexión personalizado de origen de Eventstream mediante los puertos de firewall que deben abrirse.

Creación de una secuencia de eventos

  1. Vaya al portal de Fabric.

  2. Seleccione Mi área de trabajo en la barra de navegación izquierda.

  3. En la página Mi área de trabajo, seleccione + Nuevo elemento en la barra de comandos.

  4. En la página Nuevo Elemento, busque Eventstreamy, a continuación, seleccione Eventstream.

    Captura de pantalla que muestra dónde seleccionar eventstream.

  5. En la ventana Nueva secuencia de eventos, escriba un nombre para la secuencia de eventos y luego, seleccione Crear.

    Captura de pantalla que muestra dónde crear una secuencia de eventos.

  6. La creación del nuevo flujo de eventos en el área de trabajo puede tardar unos segundos. Una vez creado el flujo de eventos, se le dirigirá al editor principal donde puede comenzar a agregar fuentes al flujo de eventos.

    Captura de pantalla que muestra la página principal de eventstream.

Obtenga la información de conexión de un origen de punto de conexión personalizado agregado.

Para obtener la información del punto de conexión de una secuencia de eventos, agregue un origen de punto de conexión personalizado a la secuencia de eventos. Luego, el punto de conexión queda disponible y expuesto en el origen del punto de conexión personalizado.

  1. Para agregar un origen de punto de conexión personalizado, en la página de introducción, seleccione Usar punto de conexión personalizado.

    Captura de pantalla que muestra cómo agregar un origen de punto de conexión personalizado.

  2. En el cuadro de diálogo Punto de conexión personalizado , escriba un nombre para el origen personalizado en Nombre de origen y, a continuación, seleccione Agregar.

    Captura de pantalla que muestra cómo escribir el nombre de origen para el punto de conexión personalizado.

  3. Después de crear el origen del punto de conexión personalizado, se agrega a la secuencia de eventos en el lienzo en modo de edición. Para implementar los datos recién agregados desde el origen de la aplicación personalizada, seleccione Publicar.

    Captura de pantalla que muestra el modo de edición de eventstream.

  4. Una vez publicada la secuencia de eventos, seleccione el icono de origen del punto de conexión personalizado en el lienzo para ver sus detalles. En el panel inferior se proporciona la información del punto de conexión necesaria para configurar la característica CES en la base de datos de Azure SQL.

    Captura de pantalla que muestra el panel de detalles del punto de conexión personalizado.

Habilitación y configuración del streaming de eventos de cambio en Azure SQL Database

Esta sección le guía a través de la habilitación y configuración del streaming de eventos de cambio para la base de datos de Azure SQL. Para empezar, elija e inicie sesión en la base de datos específica en la que desea activar esta característica y siga estos pasos:

  • Establezca la base de datos en modelo de recuperación completa si aún no está configurado.
  • Cree una clave maestra y una credencial con ámbito en la base de datos.
  • Habilite el streaming de eventos.
  • Cree el grupo de secuencias de eventos.
  • Agregue una o varias tablas al grupo de secuencias de eventos.

Hay dos protocolos que admite el streaming de eventos de cambio para transmitir los eventos de cambio a Eventstream:

  • Protocolo AMQP
  • Protocolo Kafka

Ambos enfoques requieren completar estos pasos, pero con parámetros diferentes. Además, ambas formas admiten tres tipos de autenticación: Microsoft Entra, SAS token y SAS. Puede seleccionar el protocolo y el tipo de autenticación que mejor se adapte a sus necesidades. En el ejemplo siguiente, utilizamos el protocolo AMQP y la autenticación mediante Microsoft Entra y token SAS para recorrer estos pasos. Para otros protocolos y métodos de autenticación, consulte Configuración del streaming de eventos de cambio.

1. Establezca la base de datos en el modelo de recuperación completa si aún no está configurado:

Este paso solo es necesario cuando se usa SQL Server 2025. Para Azure SQL Database o Azure SQL Managed Instance, la configuración siempre es FULL, por lo que este paso se puede omitir.

-- 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. Cree una clave maestra y una credencial con ámbito de base de datos:

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

3. Habilitar el streaming de eventos:

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

4. Cree el grupo de streaming de eventos:

Para crear el grupo de streaming de eventos, debe decidir qué autenticación se usa. Hay dos tipos de autenticación recomendados: Microsoft Entra y SAS. Para crear el grupo de flujo de eventos de modificación en una base de datos SQL mediante el protocolo AMQP, use la información clave del punto de conexión personalizado de origen de Eventstream para la configuración.

  • <EventHubNamespaceEndpoint> y <EventHubName>

    Captura de pantalla que muestra cómo obtener el nombre del centro de eventos.

  • <SharedAccessKeyName> y <PrimaryOrSecondaryKey>

    Captura de pantalla que muestra cómo obtener la clave principal del centro de eventos.

Microsoft Entra

Esta es la autenticación más segura y se recomienda encarecidamente. Siga estos pasos para configurar CES con él:

  1. Habilite la identidad administrada del sistema para la base de datos de Azure SQL en Azure portal: Security>Identity>System assigned managed identity:

    Captura de pantalla que muestra la página Identidad.

  2. Conceda permiso de Contributor o superior a la identidad administrada asignada al sistema de su base de datos de Azure SQL en el área de trabajo de Fabric donde se encuentra su secuencia de eventos: Workspace>Administrar acceso>Agregar personas o grupos, escriba el nombre de su instancia de Azure SQL para realizar la búsqueda:

    Captura de pantalla que muestra cómo buscar identidad.

  3. Inicie sesión en la base de datos de Azure SQL y ejecute los siguientes comandos SQL. Reemplace los valores de <> por los específicos de su secuencia de eventos y el entorno de 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'

Token de SAS

Si la autenticación de Microsoft Entra no está disponible, use la autenticación mediante token SAS. Siga estos pasos para configurar CES con esta autenticación:

  1. Genere un token SAS con la información del punto de conexión de su eventstream mediante un script. A continuación se muestra el ejemplo con PowerShell. Si desea usar el script en otros lenguajes, consulte Generación de un token de firma de acceso compartido. Reemplace los valores en <> con los valores proporcionados por el punto de conexión de su flujo de eventos.

    [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

  2. Inicie sesión en la base de datos de Azure SQL y ejecute los siguientes comandos SQL. Reemplace los valores de <> por los específicos de la secuencia de eventos y el entorno de 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. Agregar una o varias tablas al grupo de streaming de eventos

-- 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>'

Después de configurar el streaming de eventos de cambio en la base de datos de Azure SQL con estos pasos anteriores, puede continuar con la realización de modificaciones en la base de datos. Para confirmar que estos cambios fluyen a Eventstream, vaya a Eventstream y seleccione el nodo de secuencia predeterminado (el nodo central). Por ejemplo, se agregaron nuevos registros a la tabla o se eliminaron los registros existentes, como se muestra a continuación:

Captura de pantalla que muestra la vista en directo de la secuencia de eventos.

Extracción de datos de tabla de eventos de cambio con el operador SQL

Para extraer datos de tabla SQL de eventos de cambio, es importante comprender tanto el formato de evento como los tipos de datos de cada columna. Los eventos de cambio de SQL se adhieren a la especificación CloudEvents . Los datos de tabla reales se encuentran en la columna "data", mientras que la columna "operation" especifica si el registro se ha insertado, actualizado o eliminado. Para obtener más información, consulte Formato de mensaje JSON.

En el ejemplo siguiente se muestra cómo extraer datos de pedidos de compra de la tabla SQL "Compras" mediante el procesamiento de eventos de cambio dentro de Eventstream. A continuación, estos datos extraídos se pueden usar para analizar la información del nuevo pedido.

EXEC sp_help 'dbo.Purchases';

Captura de pantalla que muestra el resultado del aprovechamiento.

  1. Según el formato de evento de cambio descrito en formato de mensaje JSON, los datos de tabla del evento de cambio se encuentra en el nodo "eventrow" de la columna de "datos" y su tipo es "string" siguiendo el formato JSON, es decir, cadena JSON, que se puede analizar como JSON. El operador SQL de Eventstream tiene una función integrada: json_parse(string) que puede analizar la cadena JSON en un objeto JSON. Vea la captura de pantalla del ejemplo de columna de datos siguiente.

    Captura de pantalla que muestra el ejemplo de columna de datos.

  2. Puesto que solo se necesitan los datos de nuevos pedidos, solo se deben incluir los eventos de modificación con la operación "Insert", es decir, la columna "operation" con "INS": "operation" = "INS".

En el modo de edición de eventstream, agregue un "operador SQL" para extraer los datos y una secuencia derivada para recibir los datos extraídos.

Captura de pantalla que muestra cómo agregar un nodo de operador SQL.

Abra el editor de SQL en el operador SQL y escriba el lenguaje SQL como se indica a continuación:

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]

Use la prueba de consulta para obtener una vista previa de la salida de esta consulta SQL como se indica a continuación.

Captura de pantalla que muestra la salida de la consulta SQL.

Una vez que termine de configurar y publicar la secuencia de eventos, puede ver los datos de pedidos entrantes más recientes eligiendo la secuencia derivada en la vista en directo de eventstream.

Captura de pantalla que muestra la vista previa de los datos de flujo derivados.

Contenido relacionado

En este tutorial se muestra cómo transmitir eventos de cambios de SQL a Eventstream mediante el punto de conexión personalizado de origen de Eventstream con la nueva característica "Change Event Streaming" introducida recientemente en la base de datos de Azure SQL. También muestra cómo extraer los datos de tabla de los eventos de cambio con el operador SQL. Una vez que los datos llegan a eventstream y extraídos, puede enrutarlo a diferentes destinos para el análisis, las alertas y los informes. A continuación se muestran algunos recursos útiles para obtener más referencia:

  • Se actualizó por última vez el