Créer une connexion de données Event Grid pour Azure Data Explorer à l’aide de Kits de développement logiciel (SDK)

Cet article vous montre comment ingérer des objets blob de votre compte de stockage vers Azure Data Explorer en utilisant une connexion de données Event Grid. Vous allez créer une connexion de données Event Grid qui définit un abonnement Azure Event Grid. L’abonnement Event Grid route les événements de votre compte de stockage vers Azure Data Explorer via un hub d’événements Azure.

Pour savoir comment créer la connexion dans le Portail Azure ou avec un modèle ARM, consultez Créer une connexion de données Event Grid.

Pour obtenir des informations générales sur l’ingestion dans Azure Data Explorer à partir d’Event Grid, consultez Se connecter à Event Grid.

Remarque

Pour obtenir des performances optimales avec la connexion Event Grid, définissez la propriété d’ingestion rawSizeBytes via les métadonnées du blob. Pour plus d’informations, consultez Propriétés d’ingestion.

Pour obtenir des exemples de code fondés sur les versions précédentes du Kit de développement logiciel (SDK), consultez l’article archivé.

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.
• Une table de destination. Créez une table ou utilisez une table existante.
• Un mappage d’ingestion pour la table.
• Un compte de stockage. Un abonnement aux notifications Event Grid peut être défini sur des comptes de stockage Azure pour BlobStorage, StorageV2 ou Data Lake Storage Gen2.
• Enregistrez le fournisseur de ressources Event Grid.

Créer une connexion de données à Event Grid

Dans cette section, vous allez établir une connexion entre Event Grid et votre table Azure Data Explorer.

C#

1. Installez le package NuGet Microsoft.Azure.Management.Kusto.

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

3. 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 credentials = new ClientSecretCredential(tenantId, clientId, clientSecret);
   var resourceManagementClient = new ArmClient(credentials, subscriptionId);
   var resourceGroupName = "testrg";
   //The cluster and database that are created as part of the Prerequisites
   var clusterName = "mykustocluster";
   var databaseName = "mykustodatabase";
   var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
   var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
   var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
   var database = (await cluster.GetKustoDatabaseAsync(databaseName)).Value;
   var dataConnections = database.GetKustoDataConnections();
   var eventGridConnectionName = "myeventgridconnect";
   //The event hub and storage account that are created as part of the Prerequisites
   var eventHubResourceId = new ResourceIdentifier("/subscriptions/<storageAccountSubscriptionId>/resourceGroups/<storageAccountResourceGroupName>/providers/Microsoft.Storage/storageAccounts/<storageAccountName>");
   var storageAccountResourceId = new ResourceIdentifier("/subscriptions/<eventHubSubscriptionId>/resourceGroups/<eventHubResourceGroupName>/providers/Microsoft.EventHub/namespaces/<eventHubNamespaceName>/eventhubs/<eventHubName>");
   var consumerGroup = "$Default";
   var location = AzureLocation.CentralUS;
   //The table and column mapping are created as part of the Prerequisites
   var tableName = "StormEvents";
   var mappingRuleName = "StormEvents_CSV_Mapping";
   var dataFormat = KustoEventGridDataFormat.Csv;
   var blobStorageEventType = BlobStorageEventType.MicrosoftStorageBlobCreated;
   var databaseRouting = KustoDatabaseRouting.Multi;
   var eventGridConnectionData = new KustoEventGridDataConnection
   {
       StorageAccountResourceId = storageAccountResourceId, EventHubResourceId = eventHubResourceId,
       ConsumerGroup = consumerGroup, TableName = tableName, Location = location, MappingRuleName = mappingRuleName,
       DataFormat = dataFormat, BlobStorageEventType = blobStorageEventType, DatabaseRouting = databaseRouting
   };
   await dataConnections.CreateOrUpdateAsync(WaitUntil.Completed, eventGridConnectionName, eventGridConnectionData);
   

   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.
   eventGridConnectionName	myeventgridconnect	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 où la grille d’événements est configurée pour envoyer des événements.
   storageAccountResourceId	ID de ressource	ID de ressource de votre compte de stockage qui contient les données à des fins d’ingestion.
   consumerGroup	$Default	Groupe de consommateurs de votre hub d’événements.
   location	USA Centre	Emplacement de la ressource de connexion de données.
   blobStorageEventType	Microsoft.Storage.BlobCreated	Type d’événement qui déclenche l’ingestion. Les événements pris en charge sont : Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. Le renommage d’objets blob est pris en charge uniquement pour le stockage ADLSv2.
   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.

Python

1. Installez les bibliothèques nécessaires.

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

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

3. Exécutez le code suivant.

   from azure.mgmt.kusto import KustoManagementClient
   from azure.mgmt.kusto.models import EventGridDataConnection
   from azure.common.credentials import ServicePrincipalCredentials
   
   #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 = "testrg"
   #The cluster and database that are created as part of the Prerequisites
   cluster_name = "mykustocluster"
   database_name = "mykustodatabase"
   data_connection_name = "myeventhubconnect"
   #The event hub and storage account that are created as part of the Prerequisites
   event_hub_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/xxxxxx/providers/Microsoft.EventHub/namespaces/xxxxxx/eventhubs/xxxxxx"
   storage_account_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/xxxxxx/providers/Microsoft.Storage/storageAccounts/xxxxxx"
   consumer_group = "$Default"
   location = "Central US"
   #The table and column mapping that are created as part of the Prerequisites
   table_name = "StormEvents"
   mapping_rule_name = "StormEvents_CSV_Mapping"
   data_format = "csv"
   database_routing = "Multi"
   blob_storage_event_type = "Microsoft.Storage.BlobCreated"
   
   #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.begin_create_or_update(resource_group_name=resource_group_name, cluster_name=cluster_name, database_name=database_name, data_connection_name=data_connection_name,
                                               parameters=EventGridDataConnection(storage_account_resource_id=storage_account_resource_id, event_hub_resource_id=event_hub_resource_id, 
                                                                                   consumer_group=consumer_group, table_name=table_name, location=location, mapping_rule_name=mapping_rule_name, data_format=data_format, database_routing=database_routing,
                                                                                   blob_storage_event_type=blob_storage_event_type))
   # The creation of the connection is async. Validation errors are only visible if you wait for the results.
   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.
   subscription_id	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	testrg	Nom du groupe de ressources qui contient votre cluster.
   nom_cluster	mykustocluster	Nom de votre cluster.
   database_name	mykustodatabase	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	StormEvents	Nom de la table cible dans la base de données cible.
   mapping_rule_name	StormEvents_CSV_Mapping	Nom de votre mappage de colonnes associé à la table cible.
   database_routing	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.
   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 où la grille d’événements est configurée pour envoyer des événements.
   storage_account_resource_id	ID de ressource	ID de ressource de votre compte de stockage qui contient les données à des fins d’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.
   blob_storage_event_type	Microsoft.Storage.BlobCreated	Type d’événement qui déclenche l’ingestion. Les événements pris en charge sont : Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. Le renommage d’objets blob est pris en charge uniquement pour le stockage ADLSv2.

Utiliser une connexion de données Event Grid

Dans cette section, vous allez découvrir comment déclencher l’ingestion à partir de Stockage Blob Azure ou d’Azure Data Lake Gen 2 vers votre cluster après la création d’un blob ou le changement de nom d’un blob.

Sélectionnez l’onglet approprié en fonction du type de Kit de développement logiciel (SDK) de stockage utilisé pour charger des blobs.

Stockage Blob Azure

L’exemple de code suivant utilise le Kit de développement logiciel (SDK) Stockage Blob Azure pour charger un fichier dans Stockage Blob Azure. Le chargement déclenche la connexion de données Event Grid, qui ingère les données dans Azure Data Explorer.

var azureStorageAccountConnectionString=<storage_account_connection_string>;
var containerName = <container_name>;
var blobName = <blob_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mappingReference>;
// Create a new container in your storage account.
var azureStorageAccount = CloudStorageAccount.Parse(azureStorageAccountConnectionString);
var blobClient = azureStorageAccount.CreateCloudBlobClient();
var container = blobClient.GetContainerReference(containerName);
container.CreateIfNotExists();
// Set metadata and upload a file to the blob.
var blob = container.GetBlockBlobReference(blobName);
blob.Metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
blob.Metadata.Add("kustoIngestionMappingReference", mapping);
blob.UploadFromFile(localFileName);
// Confirm success of the upload by listing the blobs in your container.
var blobs = container.ListBlobs();

Remarque

Azure Data Explorer ne supprimera pas les objets blob après l’ingestion. Conservez les blobs pendant trois à cinq jours en utilisant le cycle de vie du Stockage Blob Azure pour gérer la suppression des blobs.

Azure Data Lake

L’exemple de code suivant utilise le Kit de développement logiciel (SDK) Azure Data Lake pour charger un fichier dans Data Lake Storage Gen2. Le chargement déclenche la connexion de données Event Grid, qui ingère les données dans Azure Data Explorer.

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var fileName = <file_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mapping_reference>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Create the filesystem.
var dataLakeFileSystemClient = dataLakeServiceClient.CreateFileSystem(fileSystemName).Value;
// Define file metadata and uploading options.
IDictionary<String, String> metadata = new Dictionary<string, string>();
metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
metadata.Add("kustoIngestionMappingReference", mapping);
var uploadOptions = new DataLakeFileUploadOptions
{
    Metadata = metadata,
    Close = true // Note: The close option triggers the event being processed by the data connection.
};
// Write to the file.
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(fileName);
dataLakeFileClient.Upload(localFileName, uploadOptions);

Remarque

• Lors du chargement d’un fichier avec le Kit de développement logiciel (SDK) Azure Data Lake, l’événement de création de fichier initial a une taille de 0, qui est ignorée par Azure Data Explorer pendant l’ingestion des données. Pour garantir une ingestion appropriée, définissez le paramètre Close sur true. Ce paramètre entraîne le déclenchement d’un événement FlushAndClose par la méthode de chargement. Cet événement indique que la mise à jour finale a été effectuée et que le flux de fichiers est fermé.
• Pour réduire le trafic provenant d’Event Grid ainsi que le traitement ultérieur lors de l'ingestion d'événements dans Azure Data Explorer, nous vous recommandons vivement de filtrer la clé data.api pour inclure uniquement les événements FlushAndClose. Cette opération permet de supprimer les événements de création de fichiers avec la taille 0. Pour plus d’informations sur le vidage, consultez Méthode de vidage Azure Data Lake.

Renommer des objets blob

Dans ADLSv2, il est possible de renommer des répertoires. Cependant, il est important de noter que renommer un répertoire ne déclenche pas d’événements de renommage des blobs ni ne lance l’ingestion des blobs contenus dans ce répertoire. Si vous souhaitez garantir l’ingestion de blobs après avoir renommé un répertoire, vous devez renommer directement les blob individuels dans le répertoire.

L’extrait de code suivant utilise le Kit de développement logiciel (SDK) Azure Data Lake pour renommer un blob dans un compte de stockage ADLSv2.

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var sourceFilePath = <source_file_path>;
var destinationFilePath = <destination_file_path>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Get a client to the the filesystem
var dataLakeFileSystemClient = dataLakeServiceClient.GetFileSystemClient(fileSystemName);
// Rename a file in the file system
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(sourceFilePath);
dataLakeFileClient.Rename(destinationFilePath);

Remarque

Si vous avez défini des filtres pour suivre des sujets spécifiques pendant la création de la connexion de données ou lors de la création manuelle des ressources Event Grid, ces filtres sont appliqués sur le chemin du fichier de destination.

Remarque

Le déclenchement de l’ingestion après une opération CopyBlob n’est pas pris en charge pour les comptes de stockage sur lesquels la fonctionnalité d’espace de noms hiérarchique est activée.

Important

Nous déconseillons vivement de générer des événements de stockage à partir de code personnalisé et de les envoyer à Event Hubs. Si vous choisissez de le faire, assurez-vous que les événements produits respectent strictement le schéma approprié des événements de stockage et les spécifications de format JSON.

Supprimer une connexion de données Event Grid

C#

Pour supprimer la connexion Event Grid, exécutez la commande suivante :

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

Python

Pour supprimer la connexion Event Grid, exécutez la commande suivante :

kusto_management_client.data_connections.delete(resource_group_name=resource_group_name, cluster_name=kusto_cluster_name, database_name=kusto_database_name, data_connection_name=kusto_data_connection_name)

Contenu connexe

• Traiter les données de votre instance Event Hub avec Azure Stream Analytics