Início Rápido: Enviar eventos para ou receber eventos de hubs de eventos usando Python

Neste Início Rápido, você aprenderá a enviar eventos e receber eventos de um hub de eventos usando o pacote python do azure-eventhub .

Pré-requisitos

Se você estiver conhecendo agora os Hubs de Eventos do Azure, confira Visão geral dos Hubs de Eventos antes de prosseguir com este início rápido.

Para concluir este início rápido, verifique se você tem os seguintes pré-requisitos:

• Assinatura do Microsoft Azure: inscreva-se para uma avaliação gratuita se você não tiver uma.
• Python 3.8 ou posterior: verifique se o pip está instalado e atualizado.
• Visual Studio Code (recomendado): ou use qualquer outro IDE de sua escolha.
• Namespace e hub de eventos do Event Hubs: siga este guia para criá-los no portal do Azure.

Instalar os pacotes para enviar eventos

Para instalar os pacotes do Python para Hubs de Eventos, abra um prompt de comando que tenha Python no caminho. Altere o diretório para a pasta em que você deseja manter os exemplos.

pip install azure-eventhub
pip install azure-identity
pip install aiohttp

pip install azure-eventhub

Autenticar o aplicativo no Azure

Este início rápido mostra duas maneiras de se conectar aos Hubs de Eventos do Azure:

• Sem senha. Use sua entidade de segurança no Microsoft Entra ID e o controle de acesso baseado em função (RBAC) para se conectar a um namespace do Hubs de Eventos. Você não precisa se preocupar em ter cadeias de conexão codificadas em código em seu código, em um arquivo de configuração ou em um armazenamento seguro, como o Azure Key Vault.
• Cadeia de conexão. Use uma cadeia de conexão para se conectar a um namespace dos Hubs de Eventos. Se você for iniciante no Azure, poderá achar a opção de cadeia de conexão mais fácil de seguir.

É recomendável usar a opção sem senha em aplicativos do mundo real e ambientes de produção. Para mais informações, consulte Autenticação e Autorização do Service Bus e Conexões Sem Senha para Serviços do Azure.

az eventhubs namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name>' --query id

az role assignment create --assignee "<user@___domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

New-AzRoleAssignment -SignInName <user@___domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourResourceId>

Enviar eventos

Nesta seção, crie um script Python para enviar eventos para o hub de eventos que você criou anteriormente.

1. Abra seu editor favorito do Python, como o Visual Studio Code.

2. Crie um script chamado send.py. Esse script envia um lote de eventos para o hub de eventos que você criou anteriormente.

3. Cole o seguinte código em sender.py:

   • Sem senha (recomendado)
   • Cadeia de Conexão

   No código, use valores reais para substituir os seguintes espaços reservados:

   • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE - Você pode ver o nome totalmente qualificado na página Visão geral do namespace. Ele deve estar no formato: <NAMESPACENAME>>.servicebus.windows.net.
   • EVENT_HUB_NAME - Nome do hub de eventos.

   import asyncio
   
   from azure.eventhub import EventData
   from azure.eventhub.aio import EventHubProducerClient
   from azure.identity.aio import DefaultAzureCredential
   
   EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
   EVENT_HUB_NAME = "EVENT_HUB_NAME"
   
   credential = DefaultAzureCredential()
   
   async def run():
       # Create a producer client to send messages to the event hub.
       # Specify a credential that has correct role assigned to access
       # event hubs namespace and the event hub name.
       producer = EventHubProducerClient(
           fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
           eventhub_name=EVENT_HUB_NAME,
           credential=credential,
       )
       print("Producer client created successfully.") 
       async with producer:
           # Create a batch.
           event_data_batch = await producer.create_batch()
   
           # Add events to the batch.
           event_data_batch.add(EventData("First event "))
           event_data_batch.add(EventData("Second event"))
           event_data_batch.add(EventData("Third event"))
   
           # Send the batch of events to the event hub.
           await producer.send_batch(event_data_batch)
   
           # Close credential when no longer needed.
           await credential.close()
   
   asyncio.run(run())
   

   Observação

   Para obter exemplos de outras opções para enviar eventos para um hub de eventos de forma assíncrona usando uma cadeia de conexão, consulte a página send_async.py GitHub. Os padrões mostrados também são aplicáveis ao envio de eventos sem senha.

Receber eventos

Este início rápido usa o Armazenamento de Blobs do Azure como um repositório de pontos de verificação. O repositório de pontos de verificação é usado para persistir pontos de verificação (ou seja, as últimas posições de leitura).

Siga estas recomendações ao usar o Armazenamento de Blobs do Azure como um repositório de ponto de verificação:

• Use um contêiner separado para cada grupo de consumidores. Você pode usar a mesma conta de armazenamento, mas usar um contêiner por cada grupo.
• Não use a conta de armazenamento para mais nada.
• Não use o contêiner para mais nada.
• Crie a conta de armazenamento na mesma região que o aplicativo implantado. Se o aplicativo for local, tente escolher a região mais próxima possível.

Na página Conta de armazenamento do portal do Azure, na seção serviço Blob, verifique se as seguintes configurações estão desabilitadas.

• Namespace hierárquico
• Exclusão temporária de blobs
• Controle de versão

Criar uma conta de armazenamento e um contêiner de blobs do Azure

Crie uma conta de armazenamento do Azure e um contêiner de blobs nela executando as seguintes etapas:

1. Crie uma conta de Armazenamento do Azure
2. Criar um contêiner de blob.
3. Autenticar no contêiner de blob.

Registre a cadeia de conexão e o nome do contêiner para usá-los posteriormente no código de recebimento.

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

az role assignment create --assignee "<user@___domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

New-AzRoleAssignment -SignInName <user@___domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

Instalar os pacotes para receber eventos

Para o lado receptor, será necessário instalar mais dois pacotes. Neste início rápido, você usará o Armazenamento de Blobs do Azure para persistir pontos de verificação para que o programa não leia os eventos que ele já leu. Ele executa pontos de verificação de metadados em mensagens recebidas em intervalos regulares em um blob. Essa abordagem facilita continuar a receber mensagens mais tarde exatamente do ponto em que você parou.

pip install azure-eventhub-checkpointstoreblob-aio
pip install azure-identity

pip install azure-eventhub-checkpointstoreblob-aio

Criar um script do Python para receber eventos

Nesta seção, você criará um script Python para receber eventos do hub de eventos:

1. Abra seu editor favorito do Python, como o Visual Studio Code.

2. Crie um script chamado recv.py.

3. Cole o seguinte código em recv.py:

   • Sem senha (recomendado)
   • Cadeia de Conexão

   No código, use valores reais para substituir os seguintes espaços reservados:

   • BLOB_STORAGE_ACCOUNT_URL - Esse valor deve estar no formato: https://<YOURSTORAGEACCOUNTNAME>.blob.core.windows.net/
   • BLOB_CONTAINER_NAME - Nome do contêiner de blob na conta de armazenamento do Azure.
   • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE - Você pode ver o nome totalmente qualificado na página Visão geral do namespace. Ele deve estar no formato: <NAMESPACENAME>>.servicebus.windows.net.
   • EVENT_HUB_NAME - Nome do hub de eventos.

   import asyncio
   
   from azure.eventhub.aio import EventHubConsumerClient
   from azure.eventhub.extensions.checkpointstoreblobaio import (
       BlobCheckpointStore,
   )
   from azure.identity.aio import DefaultAzureCredential
   
   BLOB_STORAGE_ACCOUNT_URL = "BLOB_STORAGE_ACCOUNT_URL"
   BLOB_CONTAINER_NAME = "BLOB_CONTAINER_NAME"
   EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
   EVENT_HUB_NAME = "EVENT_HUB_NAME"
   
   credential = DefaultAzureCredential()
   
   async def on_event(partition_context, event):
       # Print the event data.
       print(
           'Received the event: "{}" from the partition with ID: "{}"'.format(
               event.body_as_str(encoding="UTF-8"), partition_context.partition_id
           )
       )
   
       # Update the checkpoint so that the program doesn't read the events
       # that it has already read when you run it next time.
       await partition_context.update_checkpoint(event)
   
   
   async def main():
       # Create an Azure blob checkpoint store to store the checkpoints.
       checkpoint_store = BlobCheckpointStore(
           blob_account_url=BLOB_STORAGE_ACCOUNT_URL,
           container_name=BLOB_CONTAINER_NAME,
           credential=credential,
       )
   
       # Create a consumer client for the event hub.
       client = EventHubConsumerClient(
           fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
           eventhub_name=EVENT_HUB_NAME,
           consumer_group="$Default",
           checkpoint_store=checkpoint_store,
           credential=credential,
       )
       async with client:
           # Call the receive method. Read from the beginning of the partition
           # (starting_position: "-1")
           await client.receive(on_event=on_event, starting_position="-1")
   
       # Close credential when no longer needed.
       await credential.close()
   
   if __name__ == "__main__":
       # Run the main method.
       asyncio.run(main())
   

   Observação

   Para obter exemplos de outras opções para receber eventos de um hub de eventos de forma assíncrona usando uma cadeia de conexão, consulte a página recv_with_checkpoint_store_async.py GitHub. Os padrões mostrados também são aplicáveis ao recebimento de eventos sem senha.

Executar o aplicativo receptor

1. Inicie um prompt de comando.

2. Execute o comando a seguir e faça login usando a conta que foi adicionada à função de Proprietário dos Dados do Azure Event Hubs no namespace do Event Hubs e à função de Contribuidor de Dados de Blob de Armazenamento na conta de armazenamento do Azure.

   az login
   

3. Alterne para a pasta que tem o arquivo receive.py e execute o seguinte comando:

   python recv.py
   

Executar o aplicativo remetente

1. Inicie um prompt de comando.

2. Execute o comando a seguir e faça login usando a conta que foi adicionada à função de Proprietário dos Dados do Azure Event Hubs no namespace do Event Hubs e à função de Contribuidor de Dados de Blob de Armazenamento na conta de armazenamento do Azure.

   az login
   

3. Vá para a pasta que contém o send.py e então execute este comando:

   python send.py
   

A janela do destinatário deve exibir as mensagens que foram enviadas para o hub de eventos.

Solução de problemas

Se você não vir eventos na janela do receptor ou o código relatar um erro, tente as seguintes dicas de solução de problemas:

• Se você não vir os resultados de recy.py, execute send.py várias vezes.

• Se você vir erros sobre "corrotina" ao usar o código sem senha (com credenciais), verifique se está usando a importação de azure.identity.aio.

• Se você vir "Sessão de cliente não revelada" com código sem senha (com credenciais), lembre-se de fechar a credencial quando terminar. Para saber mais, confira Credenciais assíncronas.

• Se você vir erros de autorização com recv.py ao acessar o armazenamento, verifique se seguiu as etapas em Criar uma conta de armazenamento do Azure e um contêiner de blobs e atribuiu a função Colaborador de Dados de Blob de Armazenamento à entidade de serviço.

• Se você receber eventos com IDs de partição diferentes, esse resultado é esperado. As partições são um mecanismo de organização de dados relacionados ao paralelismo de downstream necessário no consumo de aplicativos. O número de partições em um hub de eventos está diretamente relacionado ao número de leitores simultâneos que você espera ter. Para obter mais informações, confira Saiba mais sobre partições.

Próximas etapas

Neste início rápido, você enviou e recebeu eventos de forma assíncrona. Para saber como enviar e receber eventos de maneira síncrona, acesse a página sync_samples do GitHub.

Explore mais exemplos e cenários avançados na biblioteca de clientes dos Hubs de Eventos do Azure para exemplos do Python.