Criar uma conexão de dados dos Hubs de Eventos para o Azure Data Explorer

Artigo
05/03/2023

O Azure Data Explorer oferece ingestão de Hubs de Eventos, uma plataforma de streaming de Big Data e serviço de ingestão de eventos. Os Hubs de Eventos podem processar milhões de eventos por segundo quase em tempo real.

Neste artigo, você se conectará a um hub de eventos e ingerirá dados no Azure Data Explorer. Para obter uma visão geral sobre a ingestão de Hubs de Eventos, consulte Conexão de dados dos Hubs de Eventos do Azure.

Pré-requisitos

Uma assinatura do Azure. Criar uma conta gratuita do Azure.
Um cluster e um banco de dados do Azure Data Explorer. Criar 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.
Um hub de eventos com dados para ingestão.

Criar uma conexão de dados do hub de eventos

Nesta seção, você estabelecerá uma conexão entre o hub de eventos e sua tabela do Azure Data Explorer. Enquanto essa conexão estiver em vigor, os dados serão transmitidos do hub de eventos para a tabela de destino. Se o hub de eventos for movido para um recurso ou assinatura diferente, você precisará atualizar ou recriar a conexão.

No portal do Azure, acesse o cluster e selecione Bancos de dados. Em seguida, selecione o banco de dados que contém sua tabela de destino.
No menu à esquerda, selecione Assimilação de dados. Em seguida, na barra superior, selecione Adicionar conexão de dados.

Preencha o formulário com as seguintes informações e, em seguida, selecione Criar.

Configuração	Valor sugerido	Descrição do campo
Nome da conexão de dados	teste de hub de conexão	O nome da conexão que você deseja criar no Azure Data Explorer.
Assinatura		A ID da assinatura em que o recurso do hub de eventos está localizado.
Namespace do Hub de Eventos	Um nome de namespace exclusivo	O nome escolhido anteriormente que identifica seu namespace.
Hub de eventos	test-hub	O hub de eventos que você criou.
Grupo de consumidores	test-group	O grupo de consumidores definido no hub de eventos que você criou.
Propriedades do sistema de eventos	Selecionar propriedades relevantes	As propriedades do sistema do hub de eventos. Se houver vários registros por mensagem de evento, as propriedades do sistema serão adicionadas ao primeiro registro. Ao adicionar propriedades do sistema, crie ou atualize o esquema de tabela e o mapeamento de maneira a incluir as propriedades selecionadas.
Compactação	Nenhum	O tipo de compactação do conteúdo das mensagens do hub de eventos. Tipos de compactação compatíveis: Nenhum, Gzip.
Identidade gerenciada (recomendável)	Atribuída pelo sistema	A identidade gerenciada usada pelo cluster do Data Explorer para acesso de leitura do hub de eventos. É recomendável usar identidades gerenciadas para controlar o acesso ao seu hub de eventos. Observação: Quando a conexão de dados é criada: * As identidades atribuídas pelo sistema serão criadas automaticamente se ainda não existirem * A identidade gerenciada é atribuída automaticamente à função receptor de dados dos Hubs de Eventos do Azure e adicionada ao cluster do Data Explorer. É recomendável verificar se a função foi atribuída e se a identidade foi adicionada ao cluster.

Observação

Se você tiver uma conexão de dados existente que não esteja usando identidades gerenciadas, é recomendável atualizá-la para usar identidades gerenciadas.

As etapas a seguir orientarão você na criação de uma conexão de hub de eventos por meio do assistente de ingestão na interface do usuário da Web do Azure Data Explorer.

Observação

Para usar o assistente, você deve ter pelo menos permissões de Usuário do Banco de Dados.
Para habilitar o acesso entre um cluster e uma conta de armazenamento sem acesso público, consulte Criar um ponto de extremidade privado gerenciado.
O cluster e o hub de eventos devem ser associados aos mesmos locatários. Caso contrário, use uma das opções do SDK, como C# ou Python.

Na guia Dados da interface do usuário da Web do Azure Data Explorer, selecione Ingerir no cartão Ingerir dados do Hub de Eventos.
A janela Ingestão de dados é aberta com a guia Destino selecionada. Os campos Cluster e Banco de Dados são preenchidos automaticamente. Você pode selecionar um cluster ou banco de dados diferente nos menus suspensos.
Em Tabela, selecione Nova tabela e insira um nome para a nova tabela. Como alternativa, use uma tabela existente.
Selecione Avançar: origem.
Em Tipo de origem, selecione Hub de Eventos.

Em Conexão de Dados, preencha os seguintes campos e selecione Avançar: Esquema.

Configuração	Valor sugerido	Descrição do campo
Assinatura		A ID da assinatura em que o recurso do hub de eventos está localizado.
Namespace do Hub de Eventos		O nome que identifica o namespace.
Hub de eventos		O hub de eventos que você deseja usar.
Nome da conexão de dados	TestDataConnection	O nome que identifica a conexão de dados.
Grupo de consumidores		O grupo de consumidores definido no hub de eventos.
Compactação		O tipo de compactação do conteúdo das mensagens do hub de eventos.
Propriedades do sistema de eventos	Selecionar propriedades relevantes	As propriedades do sistema do hub de eventos. Se houver vários registros por mensagem de evento, as propriedades do sistema serão adicionadas ao primeiro. Ao adicionar propriedades do sistema, crie ou atualize o esquema de tabela e o mapeamento de maneira a incluir as propriedades selecionadas.
Data de início da recuperação do evento	Tempo Universal Coordenado (UTC)	A conexão de dados recupera eventos existentes dos Hubs de Eventos criados após a data de início da recuperação do evento. Somente eventos retidos pelo período de retenção dos Hubs de Eventos podem ser recuperados. Se a data de início da recuperação do evento não for especificada, a hora padrão será a hora em que a conexão de dados é criada.

Defina a política de ingestão. Se o streaming estiver habilitado para o cluster, você poderá selecionar Ingestão de streaming. Se o streaming não estiver habilitado para o cluster, defina o Tempo de envio em lote. Para Hubs de Eventos, o tempo de envio em lote recomendado é de 30 segundos.
Selecione o Formato de dados. Para dados formatados em CSV, ignore o primeiro registro para ignorar a linha de cabeçalho do arquivo. Para dados formatados em JSON, selecione Ignorar erros de formato de dados para ingerir os dados no formato JSON ou deixe desmarcado para ingerir os dados no formato multijson. Selecione os níveis aninhados para determinar a divisão de dados da coluna da tabela.
Se os dados exibidos na janela de pré-visualização não estiverem completos, talvez você precise de mais dados para criar uma tabela com todos os campos de dados necessários. Use os seguintes comandos para buscar novos dados do hub de eventos:
- Descartar e buscar novos dados: descarta os dados apresentados e procura novos eventos.
- Buscar mais dados: procura mais eventos além dos eventos já encontrados.
Observação

Para ver uma pré-visualização dos dados, o hub de eventos precisa enviar os eventos.
Selecione Avançar: Resumo.
Na janela Ingestão contínua do Hub de Eventos estabelecida, todas as etapas são marcadas com marcas de seleção verdes quando o estabelecimento é concluído com êxito.

Instale o pacote NuGet Microsoft.Azure.Management.Kusto.
Crie uma entidade de aplicativo do Azure AD para usar para autenticação. Você precisará da ID do diretório (locatário), da ID do aplicativo e do segredo do cliente.

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

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.
eventHubConnectionName	myeventhubconnect	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 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.
compactação	Gzip ou Nenhum	O tipo de compactação de dados.
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 Banco de dados propriedade de ingestão. Para obter mais informações, confira Roteamento de evento.

Instale as bibliotecas necessárias.

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

Crie uma entidade de aplicativo do Azure AD para usar para autenticação. Você precisará da ID do diretório (locatário), da ID do aplicativo e do segredo do cliente.

Execute o código a seguir.

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

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.
subscriptionId	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	myresourcegroup	O nome do grupo de recursos que contém o seu cluster.
cluster_name	Meu cluster	O nome do seu cluster.
database_name	mydatabase	O nome do banco de dados de destino no cluster.
data_connection_name	myeventhubconnect	O nome desejado da conexão de dados.
table_name	mytable	O nome da tabela de destino no banco de dados de destino.
mapping_rule_name	mytablemappingrule	O nome do mapeamento de coluna relacionado à tabela de destino.
data_format	csv	O formato de dados da mensagem.
event_hub_resource_id	ID de Recurso	A ID do recurso do hub de eventos 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.
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 Banco de dados propriedade de ingestão. Para obter mais informações, confira Roteamento de evento.

O exemplo a seguir mostra um modelo do Azure Resource Manager para adicionar uma conexão de dados dos Hubs de Eventos. Você pode editar e implantar o modelo no portal do Azure usando o formulário.

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

Remover uma conexão de dados do hub de eventos

Para remover a conexão do hub de eventos do portal do Azure, faça o seguinte:

Vá para o cluster. No menu à esquerda, selecione Bancos de dados. Em seguida, selecione o banco de dados que contém a tabela de destino.
No menu à esquerda, selecione Conexões de dados. Em seguida, marque a caixa de seleção ao lado da conexão de dados do hub de eventos relevante.
Na barra de menus superior, selecione Excluir.

Para remover a conexão do hub de eventos, execute o seguinte comando:

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

Para remover a conexão do hub 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)

Próximas etapas

Consultar dados no Azure Data Explorer

Compartilhar via

Criar uma conexão de dados dos Hubs de Eventos para o Azure Data Explorer

Pré-requisitos

Criar uma conexão de dados do hub de eventos

Remover uma conexão de dados do hub de eventos

Próximas etapas

Recursos adicionais