Créer une connexion de données Event Hubs pour Azure Data Explorer

Article
05/03/2023

Azure Data Explorer offre l’ingestion à partir d’Event Hubs, une plateforme de streaming Big Data et un service d’ingestion d’événements. Event Hubs peut traiter des millions d’événements par seconde en quasi-temps réel.

Dans cet article, vous allez vous connecter à un event hub et ingérer des données dans Azure Data Explorer. Pour obtenir une vue d’ensemble de l’ingestion à partir d’Event Hubs, consultez Azure Event Hubs connexion de données.

Prérequis

Un abonnement Azure. Créez un compte Azure gratuit.
Un cluster et une base de données Azure Data Explorer. Créez un cluster et une base de données.
Table de destination. Créez une table ou utilisez une table existante.
Mappage d’ingestion pour la table.
Un hub d’événements avec des données pour l’ingestion.

Créer une connexion de données event hub

Dans cette section, vous allez établir une connexion entre le hub d’événements et votre table Azure Data Explorer. Tant que cette connexion est en place, les données sont transmises du hub d’événements à votre table cible. Si le hub d’événements est déplacé vers une autre ressource ou un autre abonnement, vous devez mettre à jour ou recréer la connexion.

Dans le Portail Azure, accédez à votre cluster et sélectionnez Bases de données. Ensuite, sélectionnez la base de données qui contient votre table cible.
Dans le menu de gauche, sélectionnez Ingestion de données. Ensuite, dans la barre supérieure, sélectionnez Ajouter une connexion de données.

Remplissez le formulaire avec les informations suivantes, puis sélectionnez Créer.

Paramètre	Valeur suggérée	Description du champ
Nom de la connexion de données	test-hub-connection	Nom de la connexion que vous souhaitez créer dans l’Explorateur de données Azure.
Abonnement		ID d’abonnement où se trouve la ressource de hub d’événements.
Espace de noms du hub d’événements	Nom unique de l’espace de noms	Nom choisi précédemment qui identifie votre espace de noms.
Event Hub	test-hub	Hub d’événements que vous avez créé.
Groupe de consommateurs	test-group	Groupe de consommateurs défini dans le hub d’événements que vous avez créé.
Propriétés du système d’événements	Sélectionner les propriétés pertinentes	Les propriétés système du hub d’événements. S’il existe plusieurs enregistrements par message d’événement, les propriétés système sont ajoutées au premier enregistrement. Lors de l’ajout des propriétés système, créez ou mettez à jour le schéma de table et le mappage pour inclure les propriétés sélectionnées.
Compression	Aucun	Type de compression de la charge utile des messages du hub d’événements. Types de compression pris en charge : Aucun, GZip.
Identité managée (recommandé)	Attribué par le système	Identité managée utilisée par le cluster Data Explorer pour l’accès en lecture à partir du hub d’événements. Nous vous recommandons d’utiliser des identités managées pour contrôler l’accès à votre hub d’événements. Remarque : Quand la connexion de données est créée : * Les identités affectées par le système sont automatiquement créées si elles n’existent pas * L’identité managée reçoit automatiquement le rôle Récepteur de données Azure Event Hubs et est ajoutée à votre cluster Data Explorer. Nous vous recommandons de vérifier que le rôle a été attribué et que l’identité a été ajoutée au cluster.

Notes

Si vous avez une connexion de données qui n’utilise pas les identités managées, nous vous recommandons de la mettre à jour pour les utiliser.

Les étapes suivantes vous guident tout au long de la création d’une connexion event hub via l’Assistant Ingestion dans l’interface utilisateur web Azure Data Explorer.

Notes

Pour utiliser l’Assistant, vous devez disposer d’au moins des autorisations utilisateur de base de données .
Pour activer l’accès entre un cluster et un compte de stockage sans accès public, consultez Créer un point de terminaison privé managé.
Le cluster et le hub d’événements doivent être associés aux mêmes locataires. Si ce n’est pas le cas, utilisez l’une des options du Kit de développement logiciel (SDK), comme C# ou Python.

Sous l’onglet Données de l’interface utilisateur web Azure Data Explorer, sélectionnez Ingérer dans le carte Ingérer des données d’Event Hub.
La fenêtre Ingérer des données s’ouvre avec l’onglet Destination sélectionné. Les champs Cluster et Base de données sont renseignés automatiquement. Vous pouvez sélectionner un autre cluster ou une autre base de données dans les menus déroulants.
Sous Table, sélectionnez Nouvelle table et entrez un nom pour la nouvelle table. Vous pouvez également utiliser une table existante.
Sélectionnez Suivant : Source.
Sous Type de source, sélectionnez Event Hub.

Sous Connexion de données, renseignez les champs suivants et sélectionnez Suivant : Schéma.

Paramètre	Valeur suggérée	Description du champ
Abonnement		ID d’abonnement où se trouve la ressource de hub d’événements.
Espace de noms du hub d’événements		Nom unique qui permet d’identifier votre espace de noms.
Event Hub		Hub d’événements que vous souhaitez utiliser.
Nom de la connexion de données	TestDataConnection	Nom qui permet d’identifier votre connexion de données.
Groupe de consommateurs		Groupe de consommateurs défini dans votre hub d’événements.
Compression		Type de compression de la charge utile des messages du hub d’événements.
Propriétés du système d’événements	Sélectionner les propriétés pertinentes	Les propriétés système du hub d’événements. S’il existe plusieurs enregistrements par message d’événement, les propriétés système sont ajoutées au premier. Lors de l’ajout des propriétés système, créez ou mettez à jour le schéma de table et le mappage pour inclure les propriétés sélectionnées.
Date de début de la récupération d’événement	temps universel (UTC, Universal Time Coordinated)	La connexion de données récupère les événements Event Hubs existants créés après la date de début de la récupération d’événements. Seuls les événements conservés par la période de rétention d’Event Hubs peuvent être récupérés. Si la date de début de la récupération d’événements n’est pas spécifiée, l’heure par défaut est l’heure à laquelle la connexion de données est créée.

Définissez la stratégie d’ingestion. Si la diffusion en continu est activée pour le cluster, vous pouvez sélectionner Ingestion de streaming. Si la diffusion en continu n’est pas activée pour le cluster, définissez l’heure de traitement par lot. Pour Event Hubs, la durée de traitement recommandée est de 30 secondes.
Sélectionnez le Format de données. Pour les données au format CSV, ignorez le premier enregistrement pour ignorer la ligne de titre du fichier. Pour les données au format JSON, sélectionnez Ignorer les erreurs de format de données pour ingérer les données au format JSON ou laissez les données non sélectionnées pour ingérer les données au format multijson. Sélectionnez les niveaux imbriqués pour déterminer la division des données de colonne de table.
Si les données que vous voyez dans la fenêtre d’aperçu sont incomplètes, il est possible que vous ayez besoin de davantage de données pour créer une table avec tous les champs de données nécessaires. Utilisez les commandes suivantes pour récupérer de nouvelles données auprès de votre hub d’événements :
- Ignorer et récupérer de nouvelles données : ignore les données présentées et recherche les nouveaux événements.
- Récupérer plus de données : recherche d’autres événements, en plus de ceux déjà trouvés.
Notes

Pour vous permettre de voir un aperçu de vos données, votre hub d’événements doit envoyer des événements.
Sélectionnez Suivant : Résumé.
Dans la fenêtre établie Ingestion continue à partir d’Event Hub, toutes les étapes sont marquées avec des marques de case activée vertes lorsque l’établissement se termine correctement.

Installez le package NuGet Microsoft.Azure.Management.Kusto.
Créez un principal d’application Azure AD à utiliser pour l’authentification. Vous aurez besoin de l’ID d’annuaire (locataire), de l’ID d’application et de la clé secrète client.

Exécutez le code suivant.

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);

Paramètre	Valeur suggérée	Description du champ
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Votre ID de client. Également appelé ID de répertoire.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID d’abonnement que vous utilisez pour la création de ressources.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID client de l’application qui peut accéder aux ressources figurant dans votre locataire.
clientSecret	PlaceholderClientSecret	Secret client de l’application qui peut accéder aux ressources figurant dans votre locataire.
resourceGroupName	testrg	Nom du groupe de ressources qui contient votre cluster.
clusterName	mykustocluster	Nom de votre cluster.
databaseName	mykustodatabase	Nom de la base de données cible dans votre cluster.
eventHubConnectionName	myeventhubconnect	Nom souhaité de votre connexion de données.
tableName	StormEvents	Nom de la table cible dans la base de données cible.
mappingRuleName	StormEvents_CSV_Mapping	Nom de votre mappage de colonnes associé à la table cible.
dataFormat	csv	Format de données du message.
eventHubResourceId	ID de ressource	ID de ressource de votre hub d’événements qui contient les données pour l’ingestion.
consumerGroup	$Default	Groupe de consommateurs de votre hub d’événements.
location	USA Centre	Emplacement de la ressource de connexion de données.
compression	Gzip ou Aucune	Type de compression de données.
databaseRouting	Multi ou Single	Routage de la base de données pour la connexion. Si vous définissez la valeur sur Single, la connexion de données est routée vers une base de données unique dans le cluster, comme spécifié dans le paramètre databaseName . Si vous définissez la valeur sur Multi, vous pouvez remplacer la base de données cible par défaut en utilisant la propriété ingestion de la base de données. Pour plus d’informations, consultez Routage des événements.

Installez les bibliothèques nécessaires.

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

Créez un principal d’application Azure AD à utiliser pour l’authentification. Vous aurez besoin de l’ID d’annuaire (locataire), de l’ID d’application et de la clé secrète client.

Exécutez le code suivant.

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())

Paramètre	Valeur suggérée	Description du champ
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Votre ID de client. Également appelé ID de répertoire.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID d’abonnement que vous utilisez pour la création de ressources.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID client de l’application qui peut accéder aux ressources figurant dans votre locataire.
client_secret	xxxxxxxxxxxxxx	Secret client de l’application qui peut accéder aux ressources figurant dans votre locataire.
resource_group_name	myresourcegroup	Nom du groupe de ressources qui contient votre cluster.
nom_cluster	mycluster	Nom de votre cluster.
database_name	ma_base_de_données	Nom de la base de données cible dans votre cluster.
data_connection_name	myeventhubconnect	Nom souhaité de votre connexion de données.
table_name	ma_table	Nom de la table cible dans la base de données cible.
mapping_rule_name	ma_règle_de_mappage_de_table	Nom de votre mappage de colonnes associé à la table cible.
data_format	csv	Format de données du message.
event_hub_resource_id	ID de ressource	ID de ressource de votre hub d’événements qui contient les données pour l’ingestion.
consumer_group	$Default	Groupe de consommateurs de votre hub d’événements.
location	USA Centre	Emplacement de la ressource de connexion de données.
databaseRouting	Multi ou Single	Routage de la base de données pour la connexion. Si vous définissez la valeur sur Single, la connexion de données est acheminée vers une base de données unique dans le cluster, comme spécifié dans le paramètre databaseName . Si vous définissez la valeur sur Multi, vous pouvez remplacer la base de données cible par défaut en utilisant la propriété ingestion de la base de données. Pour plus d’informations, consultez Routage des événements.

L’exemple suivant montre un modèle Azure Resource Manager pour ajouter une connexion de données Event Hubs. Vous pouvez modifier et déployer le modèle dans le portail Azure en utilisant le formulaire.

{
    "$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')]"
            }
        }
    ]
}

Supprimer une connexion de données event hub

Pour supprimer la connexion event hub du Portail Azure, procédez comme suit :

Accédez à votre cluster. Dans le menu de gauche, sélectionnez Bases de données. Ensuite, sélectionnez la base de données qui contient la table cible.
Dans le menu de gauche, sélectionnez Connexions de données. Ensuite, cochez la case en regard de la connexion de données du hub d’événements appropriée.
Dans la barre de menus supérieure, sélectionnez Supprimer.

Pour supprimer la connexion au hub d’événements, exécutez la commande suivante :

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