Creación de una conexión de datos de Event Hubs para Azure Data Explorer

Artículo
05/03/2023

Azure Data Explorer ofrece ingesta de Event Hubs, una plataforma de streaming de macrodatos y un servicio de ingesta de eventos. Event Hubs puede procesar millones de eventos por segundo prácticamente en tiempo real.

En este artículo, se conectará a un centro de eventos e ingerirá datos en Azure Data Explorer. Para obtener información general sobre la ingesta desde Event Hubs, consulte Azure Event Hubs conexión de datos.

Requisitos previos

Suscripción a Azure. Cree una cuenta de Azure gratuita.
Un clúster y una base de datos de Azure Data Explorer. Cree un clúster y una base de datos.
Una tabla de destino. Cree una tabla o use una tabla existente.
Asignación de ingesta para la tabla.
Un centro de eventos con datos para la ingesta.

Creación de una conexión de datos del centro de eventos

En esta sección, establecerá una conexión entre el centro de eventos y la tabla de Azure Data Explorer. Siempre que esta conexión esté en vigor, los datos se transmiten desde el centro de eventos a la tabla de destino. Si el centro de eventos se mueve a otro recurso o suscripción, deberá actualizar o volver a crear la conexión.

En el Azure Portal, vaya al clúster y seleccione Bases de datos. A continuación, seleccione la base de datos que contiene la tabla de destino.
En el menú de la izquierda, seleccione Ingesta de datos. A continuación, en la barra superior, seleccione Agregar conexión de datos.

Rellene el formulario con la siguiente información y, después, seleccione Crear.

Configuración	Valor sugerido	Descripción del campo
Nombre de la conexión de datos	test-hub-connection	Nombre de la conexión que desea crear en el Explorador de datos de Azure.
Suscripción		Identificador de la suscripción donde se encuentra el recurso del centro de eventos.
Espacio de nombres del centro de eventos	Nombre único del espacio de nombres	Nombre elegido anteriormente que identifica el espacio de nombres.
Centro de eventos	test-hub	El centro de eventos que creó.
Grupo de consumidores	test-group	Grupo de consumidores de eventos definido en el centro de eventos que creó.
Propiedades del sistema de eventos	Seleccione las propiedades pertinentes.	Propiedades del sistema del centro de eventos. Si hay varios registros por mensaje de evento, las propiedades del sistema se agregan al primer registro. Cuando agregue las propiedades del sistema, cree o actualice el esquema de tabla y la asignación para incluir las propiedades seleccionadas.
Compresión	None	Tipo de compresión de la carga de mensajes del centro de eventos. Tipos de compresión admitidos: Ninguno, Gzip.
Identidad administrada (recomendado)	Asignada por el sistema	La identidad administrada que usa el clúster de Data Explorer para el acceso de lectura del centro de eventos. Se recomienda usar identidades administradas para controlar el acceso al centro de eventos. Nota: Cuando se crea la conexión de datos: * Las identidades asignadas por el sistema se crean automáticamente si no existen * A la identidad administrada se le asigna automáticamente el rol Receptor de datos de Azure Event Hubs y se agrega al clúster de Data Explorer. Se recomienda comprobar que el rol se asignó y que la identidad se agregó al clúster.

Nota

Si tiene una conexión de datos existente que no usa identidades administradas, se recomienda actualizarla para que use identidades administradas.

Los pasos siguientes le guiarán a través de la creación de una conexión de centro de eventos a través del Asistente para ingesta en la interfaz de usuario web de Azure Data Explorer.

Nota

Para usar el asistente, debe tener al menos permisos de usuario de base de datos .
Para habilitar el acceso entre un clúster y una cuenta de almacenamiento sin acceso público, consulte Creación de un punto de conexión privado administrado.
El clúster y el centro de eventos deben estar asociados a los mismos inquilinos. Si no es así, use una de las opciones del SDK, como C# o Python.

En la pestaña Datos de la interfaz de usuario web de Azure Data Explorer, seleccione Ingest from the Ingest data from Event Hub card (Ingesta de datos desde el centro de eventos).
Se abre la ventana Ingerir datos con la pestaña Destino seleccionada. Los campos Clúster y Base de datos se rellenan automáticamente. Puede seleccionar otro clúster o base de datos en el menú desplegable.
En Tabla, seleccione Nueva tabla y escriba un nombre para la nueva tabla. También puede usar tabla existente.
Seleccione Next: Source (Siguiente: Origen).
En Tipo de origen, seleccione Centro de eventos.

En Conexión de datos, rellene los campos siguientes y seleccione Siguiente: Esquema.

Configuración	Valor sugerido	Descripción del campo
Suscripción		Identificador de la suscripción donde se encuentra el recurso del centro de eventos.
Espacio de nombres del centro de eventos		Nombre que identifica el espacio de nombres.
Centro de eventos		Centro de eventos que quiere usar.
Nombre de la conexión de datos	TestDataConnection	Nombre que identifica la conexión de datos.
Grupo de consumidores		Grupo de consumidores definido en el centro de eventos.
Compresión		Tipo de compresión de la carga de mensajes del centro de eventos.
Propiedades del sistema de eventos	Seleccione las propiedades pertinentes.	Propiedades del sistema del centro de eventos. Si hay varios registros por mensaje de evento, las propiedades del sistema se agregan a la primera. Cuando agregue las propiedades del sistema, cree o actualice el esquema de tabla y la asignación para incluir las propiedades seleccionadas.
Fecha de inicio de recuperación de eventos	Hora universal coordinada (UTC)	La conexión de datos recupera los eventos de Event Hubs existentes creados después de la fecha de inicio de la recuperación de eventos. Solo se pueden recuperar los eventos retenidos por el período de retención de Event Hubs. Si no se especifica la fecha de inicio de recuperación de eventos, la hora predeterminada es la hora en la que se crea la conexión de datos.

Establezca la directiva de ingesta. Si el streaming está habilitado para el clúster, puede seleccionar Ingesta destreaming. Si el streaming no está habilitado para el clúster, establezca el tiempo de procesamiento por lotes. Para Event Hubs, el tiempo de procesamiento por lotes recomendado es de 30 segundos.
Seleccione el formato De datos. En el caso de los datos con formato CSV, ignore el primer registro para omitir la fila de encabezado del archivo. En el caso de los datos con formato JSON, seleccione Omitir errores de formato de datos para ingerir los datos en formato JSON o dejar sin seleccionar para ingerir los datos en formato multijson. Seleccione los niveles anidados para determinar la división de datos de columna de tabla.
Si los datos que ve en la ventana de vista previa no están completos, puede que necesite más datos para crear una tabla con todos los campos de datos necesarios. Use los siguientes comandos para capturar datos nuevos desde el centro de eventos:
- Discard and fetch new data (Descartar y capturar datos nuevos): descarta los datos presentados y busca nuevos eventos.
- Fetch more data (Capturar más datos): busca más eventos además de los eventos ya encontrados.
Nota

Para ver una vista previa de los datos, el centro de eventos debe enviar eventos.
Seleccione Siguiente: Resumen.
En la ventana Ingesta continua desde el centro de eventos establecida , todos los pasos se marcan con marcas de verificación verdes cuando el establecimiento finaliza correctamente.

Instale el paquete Microsoft.Azure.Management.Kusto NuGet.
Cree una entidad de seguridad de aplicación de Azure AD que se usará para la autenticación. Necesitará el identificador de directorio (inquilino), el identificador de aplicación y el secreto de cliente.

Ejecute el código siguiente:

var tenantId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"; //Directory (tenant) ID
var clientId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"; //Application ID
var clientSecret = "PlaceholderClientSecret"; //Client Secret
var subscriptionId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx";
var authClient = ConfidentialClientApplicationBuilder.Create(clientId)
    .WithAuthority($"https://login.microsoftonline.com/{tenantId}")
    .WithClientSecret(clientSecret)
    .Build();
var result = authClient.AcquireTokenForClient(new[] { "https://management.core.windows.net/" }).ExecuteAsync().Result;
var credentials = new TokenCredentials(result.AccessToken, result.TokenType);
var kustoManagementClient = new KustoManagementClient(credentials) { SubscriptionId = subscriptionId };
var resourceGroupName = "testrg";
//The cluster and database that are created as part of the Prerequisites
var clusterName = "mykustocluster";
var databaseName = "mykustodatabase";
var eventHubConnectionName = "myeventhubconnect";
//The event hub that is created as part of the Prerequisites
var eventHubResourceId = "/subscriptions/<eventHubSubscriptionId>/resourceGroups/<eventHubResourceGroupName>/providers/Microsoft.EventHub/namespaces/<eventHubNamespaceName>/eventhubs/<eventHubName>";
var consumerGroup = "$Default";
var location = "Central US";
//The table and column mapping are created as part of the Prerequisites
var tableName = "StormEvents";
var mappingRuleName = "StormEvents_CSV_Mapping";
var dataFormat = EventHubDataFormat.CSV;
var compression = "None";
var databaseRouting = "Multi";
var eventHubConnectionData = new EventHubDataConnection(
    eventHubResourceId, consumerGroup,
    location: location, tableName: tableName, mappingRuleName: mappingRuleName,
    dataFormat: dataFormat, compression: compression, databaseRouting: databaseRouting
);
await kustoManagementClient.DataConnections.CreateOrUpdateAsync(resourceGroupName, clusterName, databaseName, eventHubConnectionName, eventHubConnectionData);

Configuración	Valor sugerido	Descripción del campo
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	El identificador de inquilino. También conocido como identificador de directorio.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identificador de suscripción que se usa para la creación de recursos.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identificador del cliente de la aplicación que puede acceder a los recursos del inquilino.
clientSecret	PlaceholderClientSecret	Secreto del cliente de la aplicación que puede acceder a los recursos del inquilino.
resourceGroupName	testrg	Nombre del grupo de recursos que contiene el clúster.
clusterName	mykustocluster	Nombre del clúster.
databaseName	mykustodatabase	Nombre de la base de datos de destino del clúster.
eventHubConnectionName	myeventhubconnect	Nombre que desea asignar a la conexión de datos.
tableName	StormEvents	Nombre de la tabla de destino de la base de datos de destino.
mappingRuleName	StormEvents_CSV_Mapping	Nombre de la asignación de columnas asociada a la tabla de destino.
dataFormat	csv	Formato de datos del mensaje.
eventHubResourceId	Identificador del recurso	Identificador del recurso del centro de eventos que contiene los datos que se van a ingerir.
consumerGroup	$Default	Grupo de consumidores del centro de eventos.
ubicación	Centro de EE. UU.	Ubicación del recurso de conexión de datos.
compression	Gzip o Ninguno	El tipo de compresión de datos.
databaseRouting	Múltiple o Sencillo	Enrutamiento de la base de datos para la conexión. Si establece el valor en Single, la conexión de datos se enruta a una base de datos única en el clúster, tal como se especifica en la configuración databaseName . Si establece el valor en Múltiple, puede invalidar la base de datos de destino predeterminada mediante la propiedad de ingestaBase de datos. Para más información, consulte Enrutamiento de eventos.

Instale las bibliotecas necesarias.

pip install azure-common
pip install azure-mgmt-kusto

Cree una entidad de seguridad de aplicación de Azure AD que se usará para la autenticación. Necesitará el identificador de directorio (inquilino), el identificador de aplicación y el secreto de cliente.

Ejecute el código siguiente:

from azure.mgmt.kusto import KustoManagementClient
from azure.mgmt.kusto.models import EventHubDataConnection
from azure.identity import ClientSecretCredential

#Directory (tenant) ID
tenant_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
#Application ID
client_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
#Client Secret
client_secret = "xxxxxxxxxxxxxx"
subscription_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
credentials = ServicePrincipalCredentials(
        client_id=client_id,
        secret=client_secret,
        tenant=tenant_id
    )
kusto_management_client = KustoManagementClient(credentials, subscription_id)

resource_group_name = "myresourcegroup"
#The cluster and database that are created as part of the Prerequisites
cluster_name = "mycluster"
database_name = "mydatabase"
data_connection_name = "myeventhubconnect"
#The event hub that is created as part of the Prerequisites
event_hub_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/myresourcegroup/providers/Microsoft.EventHub/namespaces/myeventhubnamespace/eventhubs/myeventhub"";
consumer_group = "$Default"
location = "Central US"
#The table and column mapping that are created as part of the Prerequisites
table_name = "mytable"
mapping_rule_name = "mytablemappingrule"
data_format = "csv"
database_routing = "Multi"
#Returns an instance of LROPoller, check https://learn.microsoft.com/python/api/msrest/msrest.polling.lropoller?view=azure-python
poller = kusto_management_client.data_connections.create_or_update(
            resource_group_name=resource_group_name,
            cluster_name=cluster_name,
            database_name=database_name,
            data_connection_name=data_connection_name,
            parameters=EventHubDataConnection(
                event_hub_resource_id=event_hub_resource_id,
                consumer_group=consumer_group,
                location=location,
                table_name=table_name,
                mapping_rule_name=mapping_rule_name,
                data_format=data_format,
                database_routing=database_routing
            )
        )
poller.wait()
print(poller.result())

Configuración	Valor sugerido	Descripción del campo
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	El identificador de inquilino. También conocido como identificador de directorio.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identificador de suscripción que se usa para la creación de recursos.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identificador del cliente de la aplicación que puede acceder a los recursos del inquilino.
client_secret	xxxxxxxxxxxxxx	Secreto del cliente de la aplicación que puede acceder a los recursos del inquilino.
resource_group_name	myresourcegroup	Nombre del grupo de recursos que contiene el clúster.
cluster_name	mycluster	Nombre del clúster.
database_name	mydatabase	Nombre de la base de datos de destino del clúster.
data_connection_name	myeventhubconnect	Nombre que desea asignar a la conexión de datos.
table_name	mytable	Nombre de la tabla de destino de la base de datos de destino.
mapping_rule_name	mytablemappingrule	Nombre de la asignación de columnas asociada a la tabla de destino.
data_format	csv	Formato de datos del mensaje.
event_hub_resource_id	Identificador del recurso	Identificador del recurso del centro de eventos que contiene los datos que se van a ingerir.
consumer_group	$Default	Grupo de consumidores del centro de eventos.
ubicación	Centro de EE. UU.	Ubicación del recurso de conexión de datos.
databaseRouting	Múltiple o Sencillo	Enrutamiento de la base de datos para la conexión. Si establece el valor en Single, la conexión de datos se enruta a una base de datos única en el clúster, tal como se especifica en la configuración databaseName . Si establece el valor en Múltiple, puede invalidar la base de datos de destino predeterminada mediante la propiedad de ingestaBase de datos. Para más información, consulte Enrutamiento de eventos.

En el ejemplo siguiente se muestra una plantilla de Azure Resource Manager para agregar una conexión de datos de Event Hubs. Puede editar e implementar la plantilla en Azure Portal mediante el formulario.

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "namespaces_eventhubns_name": {
            "type": "string",
            "defaultValue": "eventhubns",
            "metadata": {
                "description": "Specifies the Event Hubs Namespace name."
            }
        },
        "EventHubs_eventhubdemo_name": {
            "type": "string",
            "defaultValue": "eventhubdemo",
            "metadata": {
                "description": "Specifies the event hub name."
            }
        },
        "consumergroup_default_name": {
            "type": "string",
            "defaultValue": "$Default",
            "metadata": {
                "description": "Specifies the consumer group of the event hub."
            }
        },
        "Clusters_kustocluster_name": {
            "type": "string",
            "defaultValue": "kustocluster",
            "metadata": {
                "description": "Specifies the name of the cluster"
            }
        },
        "databases_kustodb_name": {
            "type": "string",
            "defaultValue": "kustodb",
            "metadata": {
                "description": "Specifies the name of the database"
            }
        },
        "tables_kustotable_name": {
            "type": "string",
            "defaultValue": "kustotable",
            "metadata": {
                "description": "Specifies the name of the table"
            }
        },
        "mapping_kustomapping_name": {
            "type": "string",
            "defaultValue": "kustomapping",
            "metadata": {
                "description": "Specifies the name of the mapping rule"
            }
        },
        "dataformat_type": {
            "type": "string",
            "defaultValue": "csv",
            "metadata": {
                "description": "Specifies the data format"
            }
        },
        "databaseRouting_type": {
            "type": "string",
            "defaultValue": "Single",
            "metadata": {
                "description": "The database routing for the connection. If you set the value to **Single**, the data connection will be routed to a single database in the cluster as specified in the *databaseName* setting. If you set the value to **Multi**, you can override the default target database using the *Database* EventData property."
            }
        },
        "dataconnections_kustodc_name": {
            "type": "string",
            "defaultValue": "kustodc",
            "metadata": {
                "description": "Name of the data connection to create"
            }
        },
        "subscriptionId": {
            "type": "string",
            "defaultValue": "[subscription().subscriptionId]",
            "metadata": {
                "description": "Specifies the subscriptionId of the event hub"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the event hub"
            }
        },
        "location": {
            "type": "string",
            "defaultValue": "[resourceGroup().location]",
            "metadata": {
                "description": "Location for all resources."
            }
        }
    },
    "variables": {
    },
    "resources": [{
            "type": "Microsoft.Kusto/Clusters/Databases/DataConnections",
            "apiVersion": "2022-02-01",
            "name": "[concat(parameters('Clusters_kustocluster_name'), '/', parameters('databases_kustodb_name'), '/', parameters('dataconnections_kustodc_name'))]",
            "location": "[parameters('location')]",
            "kind": "EventHub",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "eventHubResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.EventHub/namespaces/eventhubs', parameters('namespaces_eventhubns_name'), parameters('EventHubs_eventhubdemo_name'))]",
                "consumerGroup": "[parameters('consumergroup_default_name')]",
                "tableName": "[parameters('tables_kustotable_name')]",
                "mappingRuleName": "[parameters('mapping_kustomapping_name')]",
                "dataFormat": "[parameters('dataformat_type')]",
                "databaseRouting": "[parameters('databaseRouting_type')]"
            }
        }
    ]
}

Eliminación de una conexión de datos del centro de eventos

Para quitar la conexión del centro de eventos de la Azure Portal, haga lo siguiente:

Vaya al clúster. En el menú de la izquierda, seleccione Bases de datos. A continuación, seleccione la base de datos que contiene la tabla de destino.
En el menú de la izquierda, seleccione Conexiones de datos. A continuación, active la casilla situada junto a la conexión de datos del centro de eventos correspondiente.
En la barra de menús superior, seleccione Eliminar.

Para quitar la conexión del centro de eventos, ejecute el siguiente comando:

kustoManagementClient.DataConnections.Delete(resourceGroupName, clusterName, databaseName, dataConnectionName);