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

Neste artigo, você aprende a incluir blobs de sua conta de armazenamento no Azure Data Explorer usando uma conexão de dados de grade de eventos. Você criará uma conexão de dados de grade de eventos que define uma assinatura da Grade de Eventos do Azure. A assinatura da Grade de Eventos roteia eventos de sua conta de armazenamento para o Azure Data Explorer por meio dos Hubs de Eventos do Azure.

Para saber como criar a conexão no portal do Azure ou com um modelo do ARM, consulte Criar uma conexão de dados da Grade de Eventos.

Para obter informações gerais sobre a ingestão no Azure Data Explorer da Grade de Eventos, confira Conectar-se à Grade de Eventos.

Observação

Para obter o melhor desempenho com a conexão da Grade de Eventos, defina a rawSizeBytes propriedade de ingestão por meio dos metadados de blob. Para obter mais informações, consulte propriedades de ingestão.

Para obter exemplos de código com base em versões anteriores do SDK, consulte o artigo arquivado.

Pré-requisitos

• Uma assinatura do Azure. Criar uma conta gratuita do Azure.
• Um cluster e um banco de dados do Azure Data Explorer. Crie um cluster e um banco de dados.
• Uma tabela de destino. Crie uma tabela ou use uma tabela existente.
• Um mapeamento de ingestão para a tabela.
• Uma conta de armazenamento. Uma assinatura de notificação da Grade de Eventos pode ser definida em contas de Armazenamento do Azure para BlobStorage, StorageV2ou Data Lake Storage Gen2.
• Registre o provedor de recursos da Grade de Eventos.

Criar uma conexão de dados da Grade de Eventos

Nesta seção, você estabelecerá uma conexão entre a Grade de Eventos e a tabela Data Explorer do Azure.

C#

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

2. Crie um Microsoft Entra entidade de aplicativo a ser usada para autenticação. Você precisará da ID do diretório (locatário), da ID do aplicativo e do segredo 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);
   

   Configuração	Valor sugerido	Descrição do campo
   tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID do locatário. Também conhecida como ID do diretório.
   subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID da 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 seu cluster.
   clusterName	mykustocluster	O nome do seu cluster.
   databaseName	mykustodatabase	O nome do banco de dados de destino no cluster.
   eventGridConnectionName	myeventgridconnect	O nome desejado da conexão de dados.
   tableName	StormEvents	O nome da tabela de destino no banco de dados de destino.
   mappingRuleName	StormEvents_CSV_Mapping	O nome do mapeamento de coluna relacionado à tabela de destino.
   dataFormat	csv	O formato de dados da mensagem.
   eventHubResourceId	ID de Recurso	A ID do recurso do hub de eventos em que a Grade de Eventos foi configurada para enviar eventos.
   storageAccountResourceId	ID de Recurso	A ID do recurso da conta de armazenamento que contém os dados para ingestão.
   consumerGroup	$Default	O grupo de consumidores do hub de eventos.
   local	Centro dos EUA	A localização do recurso de conexão de dados.
   blobStorageEventType	Microsoft.Storage.BlobCreated	O tipo de evento que dispara a ingestão. Os eventos com suporte são: Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. Há suporte para renomear blobs somente no armazenamento de ADLSv2.
   databaseRouting	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 banco de dados individual 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 de ingestãoDatabase. Para obter mais informações, confira Roteamento de evento.

Python

1. Instale as bibliotecas necessárias.

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

2. Crie um Microsoft Entra entidade de aplicativo a ser usada para autenticação. Você precisará da ID do diretório (locatário), da 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())
   

   Configuração	Valor sugerido	Descrição do campo
   tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID do locatário. Também conhecida como ID do diretório.
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	A ID da assinatura que você usa para a criação de recursos.
   client_id	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.
   resource_group_name	testrg	O nome do grupo de recursos que contém o seu cluster.
   cluster_name	mykustocluster	O nome do seu cluster.
   database_name	mykustodatabase	O nome do banco de dados de destino no cluster.
   data_connection_name	myeventhubconnect	O nome desejado da conexão de dados.
   table_name	StormEvents	O nome da tabela de destino no banco de dados de destino.
   mapping_rule_name	StormEvents_CSV_Mapping	O nome do mapeamento de coluna 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 banco de dados individual 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 de ingestãoDatabase. Para obter mais informações, confira Roteamento de evento.
   data_format	csv	O formato de dados da mensagem.
   event_hub_resource_id	ID de Recurso	A ID do recurso do hub de eventos em que a Grade de Eventos foi configurada para enviar eventos.
   storage_account_resource_id	ID de Recurso	A ID do recurso da conta de armazenamento que contém os dados para ingestão.
   consumer_group	$Default	O grupo de consumidores do hub de eventos.
   local	Centro dos EUA	A localização do recurso de conexão de dados.
   blob_storage_event_type	Microsoft.Storage.BlobCreated	O tipo de evento que dispara a ingestão. Os eventos com suporte são: Microsoft.Storage.BlobCreated ou Microsoft.Storage.BlobRenamed. Há suporte para renomear blobs somente no armazenamento de ADLSv2.

Usar a conexão de dados da Grade de Eventos

Esta seção mostra como disparar a ingestão do Armazenamento de Blobs do Azure ou do Azure Data Lake Gen 2 para o cluster após a criação de blobs ou renomeação de blob.

Selecione a guia relevante com base no tipo de SDK de armazenamento usado para carregar blobs.

Armazenamento de Blobs do Azure

O exemplo de código a seguir usa o SDK do Armazenamento de Blobs do Azure para carregar um arquivo no Armazenamento de Blobs do Azure. O upload dispara 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();

Observação

O Azure Data Explorer não excluirá os blobs após a ingestão. Mantenha os blobs por três a cinco dias usando o ciclo de vida do Armazenamento de Blobs do Azure para gerenciar a exclusão de blobs.

Azure Data Lake

O exemplo de código a seguir usa o SDK do Azure Data Lake para carregar um arquivo para Data Lake Storage Gen2. O upload dispara 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);

Observação

• Ao carregar um arquivo com o SDK do Azure Data Lake, o evento de criação de arquivo inicial tem um tamanho de 0, que é ignorado pelo Azure Data Explorer durante a ingestão de dados. Para garantir a ingestão adequada, defina o Close parâmetro como true. Esse parâmetro faz com que o método de upload dispare um evento FlushAndClose , indicando que a atualização final foi feita e o fluxo de arquivos está fechado.
• Para reduzir o tráfego proveniente da Grade de Eventos e do processamento subsequente ao ingerir eventos no Azure Data Explorer, recomendamos filtrar a chave data.api para incluir apenas eventos FlushAndClose, removendo assim eventos de criação de arquivo com tamanho 0. Para obter mais informações sobre a liberação, confira Método de liberação do Azure Data Lake.

Renomear blobs

No ADLSv2, é possível renomear diretórios. No entanto, é importante observar que renomear um diretório não dispara eventos renomeados por blob nem inicia a ingestão de blobs contidos no diretório. Se você quiser garantir a ingestão de blobs depois de renomear um diretório, renomeie 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);

Observação

Se você definiu filtros para acompanhar assuntos específicos ao criar a conexão de dados ou ao criar os recursos da Grade de Eventos manualmente, esses filtros serão aplicados no caminho do arquivo de destino.

Observação

Não há suporte para disparar a ingestão após uma CopyBlob operação para contas de armazenamento que têm o recurso de namespace hierárquico habilitado.

Remover uma conexão de dados da Grade de Eventos

C#

Para remover a conexão da Grade de Eventos, execute o seguinte comando:

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

Python

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)

Conteúdo relacionado

• Processar dados do hub de eventos usando o Azure Stream Analytics