Criar uma conexão de dados da Grade de Eventos para o Azure Data Explorer com SDKs

1. Instale o pacote Microsoft.Azure.Management.Kusto NuGet.

2. Crie uma entidade de aplicação Microsoft Entra para utilizar para autenticação. Você precisará do ID do diretório (locatário), do ID do aplicativo e da chave secreta do cliente.

3. Execute o código a seguir.

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

   Setting	Valor sugerido	Descrição do campo
   tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	O ID do inquilino. Também conhecido como ID de diretório.
   ID de subscrição	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID de assinatura que você usa para a criação de recursos.
   clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID do cliente do aplicativo que pode acessar recursos em seu locatário.
   clientSecret	PlaceholderClientSecret	O segredo do cliente do aplicativo que pode acessar recursos em seu locatário.
   resourceGroupName	Testrg	O nome do grupo de recursos que contém o cluster.
   nome_do_cluster	Mykustocluster	O nome do seu cluster.
   databaseName	mykustodatabase	O nome do banco de dados de destino no cluster.
   eventGridConnectionName	MyEventGridConnect	O nome desejado da sua conexão de dados.
   tableName	StormEvents	O nome da tabela de destino no banco de dados de destino.
   nomeDaRegraDeMapeamento	StormEvents_CSV_Mapping	O nome do mapeamento de colunas relacionado à tabela de destino.
   formato de dados	CSV	O formato de dados da mensagem.
   ID do recurso do Event Hub	ID do recurso	O ID do recurso do hub de eventos onde o Event Grid está configurado para enviar eventos.
   storageAccountResourceId	ID do recurso	O ID do recurso da sua conta de armazenamento que contém os dados para ingestão.
   Grupo de consumidores	$Default	O grupo de consumidores do seu hub de eventos.
   localização	Centro dos EUA	O local do recurso de conexão de dados.
   blobStorageEventType	Microsoft.Storage.BlobCreated	O tipo de evento que desencadeia a ingestão. Os eventos suportados são: Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. A renomeação de Blob é suportada apenas para armazenamento ADLSv2.
   roteamento de base de dados	Multi ou Single	O roteamento de banco de dados para a conexão. Se você definir o valor como Single, a conexão de dados será roteada para um único banco de dados no cluster, conforme especificado na configuração databaseName . Se você definir o valor como Multi, poderá substituir o banco de dados de destino padrão usando a propriedade Databaseingestion. Para obter mais informações, consulte Roteamento de eventos.

1. Instale as bibliotecas necessárias.

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

2. Crie uma entidade de aplicação Microsoft Entra para utilizar para autenticação. Você precisará do ID do diretório (locatário), do ID do aplicativo e do segredo do cliente.

3. Execute o código a seguir.

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

   Setting	Valor sugerido	Descrição do campo
   ID do inquilino	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	O ID do inquilino. Também conhecido como ID de diretório.
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID de assinatura que você usa para a criação de recursos.
   ID do cliente	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID do cliente do aplicativo que pode acessar recursos em seu locatário.
   client_secret	xxxxxxxxxxxxxx	O segredo do cliente do aplicativo que pode acessar recursos em seu locatário.
   nome_grupo_recursos	Testrg	O nome do grupo de recursos que contém o cluster.
   nome do cluster	Mykustocluster	O nome do seu cluster.
   nome_da_base_de_dados	mykustodatabase	O nome do banco de dados de destino no cluster.
   nome_da_conexão_de_dados	MyEventHubConnect	O nome desejado da sua conexão de dados.
   nome_da_tabela	StormEvents	O nome da tabela de destino no banco de dados de destino.
   nome_da_regra_de_mapeamento	StormEvents_CSV_Mapping	O nome do mapeamento de colunas relacionado à tabela de destino.
   database_routing	Multi ou Single	O roteamento de banco de dados para a conexão. Se você definir o valor como Single, a conexão de dados será roteada para um único banco de dados no cluster, conforme especificado na configuração databaseName . Se você definir o valor como Multi, poderá substituir o banco de dados de destino padrão usando a propriedade Databaseingestion. Para obter mais informações, consulte Roteamento de eventos.
   formato_de_dados	CSV	O formato de dados da mensagem.
   Identificador_de_recurso_do_event_hub	ID do recurso	O ID do recurso do hub de eventos onde o Event Grid está configurado para enviar eventos.
   identificador_recurso_conta_armazenamento	ID do recurso	O ID do recurso da sua conta de armazenamento que contém os dados para ingestão.
   grupo de consumidores	$Default	O grupo de consumidores do seu hub de eventos.
   localização	Centro dos EUA	O local do recurso de conexão de dados.
   blob_storage_tipo_de_evento	Microsoft.Storage.BlobCreated	O tipo de evento que desencadeia a ingestão. Os eventos suportados são: Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. A renomeação de Blob é suportada apenas para armazenamento ADLSv2.

O exemplo de código a seguir usa o SDK de Armazenamento de Blob do Azure para carregar um arquivo no Armazenamento de Blobs do Azure. O carregamento aciona a conexão de dados da Grade de Eventos, que ingere os dados no 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();

O exemplo de código a seguir usa o SDK do Azure Data Lake para carregar um arquivo no Data Lake Storage Gen2. O carregamento aciona a conexão de dados da Grade de Eventos, que ingere os dados no 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);

Renomear objetos blob

No ADLSv2, é possível renomear diretórios. No entanto, é importante observar que renomear um diretório não aciona eventos de renomeação de blobs ou inicia a ingestão de blobs contidos no diretório. Se quiser garantir a ingestão de blobs depois de renomear um diretório, você deve renomear diretamente os blobs individuais dentro do diretório.

O exemplo de código a seguir mostra como renomear um blob em uma conta de armazenamento 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);

Para remover a conexão da grade de eventos, execute o seguinte comando:

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

Para remover a conexão da grade de eventos, execute o seguinte comando:

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)